options-automation/trades_table.py

from pandas import DataFrame

from database.trades import get_leg, get_spread, trade, upsert
from database.trades.action import Action
from database.trades.option_type import OptionType
from ibkr import option_type as IBKROptionType
from ibkr.option_order import OptionOrder

from iron_condor_trade import IronCondorTrade

def translate_action(action_string: str) -> Action:
    for action in Action:
        if action_string == action.value:
            return action

def translate_option_type(option_type_string: str) -> OptionType:
    if option_type_string == IBKROptionType.CALL:
        return OptionType.CALL
    else:
        return OptionType.PUT

def spread(spread_order: OptionOrder) -> dict:
    near_leg = spread_order.legs[0]
    far_leg = spread_order.legs[1]
    return get_spread(
        near_leg = get_leg(
            action = translate_action(near_leg.action),
            strike = near_leg.strike,
            option_type = translate_option_type(near_leg.option_type)
        ),
        far_leg = get_leg(
            action = translate_action(far_leg.action),
            strike = far_leg.strike,
            option_type = translate_option_type(far_leg.option_type)
        ),
        open_price = spread_order.fill_price,
        entry_slippage = round(spread_order.mid_price - spread_order.fill_price, 3)
    )

def insert_trade(iron_condor: IronCondorTrade, call_spread_order: OptionOrder, put_spread_order: OptionOrder):
    upsert(
        DataFrame([{
            'Date': iron_condor.entry_time.date(),
            'Symbol': iron_condor.symbol,
            'Strategy': iron_condor.strategy,
            'Entry Time': iron_condor.entry_time,
            'Exit Time': None, # Required.
            'Spreads': [spread(call_spread_order), spread(put_spread_order)],
            'Profit': None # Required.
        }])
    )

def update_trade(iron_condor: IronCondorTrade, stop_order: OptionOrder):
    date = iron_condor.entry_time.date()

    trade_record = trade(
        date = date, 
        symbol = iron_condor.symbol, 
        strategy = iron_condor.strategy, 
        entry_time = iron_condor.entry_time
    )
    spreads = trade_record['Spreads'].iloc[0]

    option_type = translate_option_type(stop_order.legs[0].option_type)
    spread_index = 0 if (option_type == OptionType.CALL) else 1

    spreads[spread_index]['Close'] = stop_order.fill_price

    stop_price = spreads[spread_index]['Open'] * iron_condor.stop_multiple
    spreads[spread_index]['Exit Slippage'] = round(stop_order.fill_price - stop_price, 3)

    upsert(
        DataFrame([{
            'Date': date,
            'Symbol': iron_condor.symbol,
            'Strategy': iron_condor.strategy,
            'Entry Time': iron_condor.entry_time,  
            'Exit Time': None, 
            'Spreads': spreads,
            'Profit': None # To be updated end of day.
        }])
    )