options-automation/iron_condor_scheduler.py

import schedule
import time

from backtesting.filter import VVIXRegimeFilter
from datetime import datetime
from dotenv import load_dotenv
from ibkr import Client
from iron_condor import enter_iron_condor
from multiprocessing import Process
from os import getenv
from pytz import timezone

load_dotenv()

eastern_timezone = 'America/New_York'

trade_filters = [VVIXRegimeFilter()]

def enter_trade(entry_time: datetime):
    job_process = Process(target = enter_iron_condor, args = [entry_time])
    job_process.start()

def connection_successful():
    try:
        Client()
        return True
    except Exception:
        return False

if __name__ == '__main__':
    assert connection_successful(), 'Cannot connect to IBKR. Ensure that TWS or Gateway is running.'

    now = datetime.now(timezone(eastern_timezone))
    assert all(filter.trade_allowed(now) for filter in trade_filters), 'Trade filters prevent trading today.'

    entry_times = getenv('ENTRY_TIMES').split(',')
    for entry_time in entry_times:
        schedule_time = datetime.strptime(entry_time, '%H:%M:%S').replace(
            year = now.year, 
            month = now.month, 
            day = now.day,
            tzinfo = now.tzinfo
        )

        # Prevent scheduling for times that have already elapsed.
        if schedule_time > now:
            print(f'Scheduling for {entry_time}.')
            schedule.every().day.at(entry_time, eastern_timezone).do(enter_trade, schedule_time)

    while True:
        schedule.run_pending()
        time.sleep(1)