backtest-automation/download_spx_quotes.py

import os
import pandas as pd
import paramiko
import schedule
import time
import zipfile

from dotenv import load_dotenv
from os import getenv

load_dotenv()

quotes_directory = getenv('DOWNLOAD_DIRECTORY')
download_directory = quotes_directory + 'Zip Archive/'

def process_file(file: str):
    quote_data = pd.read_csv(file)
    current_date = quote_data.iloc[0]['quote_datetime'][:10]
    zero_dte_quotes = quote_data[(quote_data['expiration'] == current_date) & (quote_data['root'] == 'SPXW')]
    if not zero_dte_quotes.empty:
        zero_dte_quotes.to_csv(os.path.join(quotes_directory, current_date[:4], current_date + '.csv'), index=False)

def download_quotes():
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect(
        hostname='sftp.datashop.livevol.com', 
        username=getenv('USER'), 
        password=getenv('PASSWORD')
    )
    sftp = client.open_sftp()
    sftp.chdir('./subscriptions/' + getenv('ORDER') + '/' + getenv('ITEM'))
    files = sftp.listdir()
    for file in files:
        local_file = download_directory + file
        if not os.path.exists(local_file):
            print('Downloading', file)
            sftp.get(file, local_file)
            with zipfile.ZipFile(local_file, 'r') as zip_file:
                print('Extracting', file)
                zip_file.extractall(download_directory)
                extracted_file = local_file.replace('.zip', '.csv')
                print('Processing', extracted_file)
                process_file(extracted_file)
                print('Deleting', extracted_file)
                os.remove(extracted_file)
    client.close()

if __name__ == '__main__':
    schedule.every().day.at('04:00', 'America/Los_Angeles').do(download_quotes)
    while True:
        schedule.run_pending()
        time.sleep(1)