Refactor strategy creation to enable clients to more easily perform backtests

This commit is contained in:
moshferatu 2024-01-17 11:37:37 -08:00
parent 93930bfea8
commit 02763460cf
7 changed files with 105 additions and 85 deletions

View File

@ -0,0 +1,38 @@
from backtesting import backtest_iron_condor, DeltaTargetStrategy, OptionType
from datetime import datetime
def create_strategies(entry_time: str, number_of_contracts: int = 1):
call_spread_strategy = DeltaTargetStrategy(
delta_upper_bound = 0.11,
delta_lower_bound = 0.10,
option_type = OptionType.CALL,
number_of_contracts = number_of_contracts,
spread_width = 50,
stop_loss_multiple = 1.00,
trade_entry_time = entry_time
)
put_spread_strategy = DeltaTargetStrategy(
delta_upper_bound = 0.11,
delta_lower_bound = 0.10,
option_type = OptionType.PUT,
number_of_contracts = number_of_contracts,
spread_width = 50,
stop_loss_multiple = 1.00,
trade_entry_time = entry_time
)
return call_spread_strategy, put_spread_strategy
if __name__ == '__main__':
start_date = datetime(2024, 1, 12)
end_date = datetime.now()
call_spread_strategy, put_spread_strategy = create_strategies(entry_time = '10:05:00')
backtest_result = backtest_iron_condor(
f'Iron Condor @ {call_spread_strategy.trade_entry_time}',
call_spread_strategy,
put_spread_strategy,
start_date,
end_date
)
print(backtest_result)
# TODO: Move plot() to plotting module.
# plot(backtest_result, title = 'Iron Condor Backtest Results')

View File

@ -1 +1,5 @@
from backtest_iron_condor import backtest_iron_condor
from .backtest_iron_condor import backtest_iron_condor
from .credit_target_strategy import CreditTargetStrategy
from .delta_target_strategy import DeltaTargetStrategy
from .option_spread_strategy import OptionSpreadStrategy
from .option_type import OptionType

View File

@ -4,10 +4,14 @@ import os
import pandas as pd
import plotly.express as px
from dotenv import load_dotenv
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import Enum
from datetime import datetime
from dotenv import load_dotenv
from .credit_target_strategy import CreditTargetStrategy
from .delta_target_strategy import DeltaTargetStrategy
from .option_spread_strategy import OptionSpreadStrategy
from .option_type import OptionType
load_dotenv()
@ -19,23 +23,6 @@ MARKET_OPEN = '09:35:00'
OPTION_DATA_DIRECTORY = os.getenv('OPTION_DATA_DIRECTORY')
STRIKE_MULTIPLE = 5.0
class OptionType(Enum):
PUT = 'P'
CALL = 'C'
@dataclass
class OptionStrat:
delta_upper_bound: float
delta_lower_bound: float
credit_target: float
max_loss: float
number_of_contracts: int
option_type: OptionType
spread_width: int
stop_loss_multiple: float
stop_loss_percent: float
trade_entry_time: str
@dataclass
class BacktestResult:
date: str
@ -92,7 +79,7 @@ def plot(backtest_results: pd.DataFrame, title: str):
chart.update_xaxes(rangebreaks=[dict(values=list(excluded_dates))])
chart.show()
def get_spread_history_credit(historical_option_data: pd.DataFrame, option_strat: OptionStrat) -> pd.DataFrame:
def get_spread_history_credit(historical_option_data: pd.DataFrame, option_strat: CreditTargetStrategy) -> pd.DataFrame:
current_date = historical_option_data.iloc[0]['quote_datetime'][:10]
opening_quotes = historical_option_data[(historical_option_data['quote_datetime'] == (current_date + ' ' + option_strat.trade_entry_time))]
if opening_quotes.empty:
@ -126,7 +113,7 @@ def get_spread_history_credit(historical_option_data: pd.DataFrame, option_strat
spread_history = short_strike_history.join(long_strike_history, lsuffix='_short_strike', rsuffix='_long_strike', on='quote_datetime')
return spread_history
def get_spread_history(historical_option_data: pd.DataFrame, option_strat: OptionStrat) -> pd.DataFrame:
def get_spread_history(historical_option_data: pd.DataFrame, option_strat: DeltaTargetStrategy) -> pd.DataFrame:
current_date = historical_option_data.iloc[0]['quote_datetime'][:10]
opening_quotes = historical_option_data[(historical_option_data['quote_datetime'] == (current_date + ' ' + option_strat.trade_entry_time))]
if opening_quotes.empty:
@ -163,16 +150,16 @@ def get_spread_history(historical_option_data: pd.DataFrame, option_strat: Optio
def _backtest_iron_condor(
historical_option_data: pd.DataFrame,
call_spread_strat: OptionStrat,
put_spread_strat: OptionStrat
call_spread_strategy: OptionSpreadStrategy,
put_spread_strategy: OptionSpreadStrategy
) -> BacktestResult:
call_spread_history = get_spread_history(historical_option_data, call_spread_strat)
put_spread_history = get_spread_history(historical_option_data, put_spread_strat)
call_spread_history = get_spread_history(historical_option_data, call_spread_strategy)
put_spread_history = get_spread_history(historical_option_data, put_spread_strategy)
current_date = call_spread_history.iloc[0].name[:10]
entry_time = call_spread_strat.trade_entry_time
entry_time = call_spread_strategy.trade_entry_time
call_spread_entry = call_spread_history.loc[current_date + ' ' + entry_time]
original_call_spread_price = ((call_spread_entry['ask_short_strike'] + call_spread_entry['bid_short_strike']) / 2.0) - ((call_spread_entry['ask_long_strike'] + call_spread_entry['bid_long_strike']) / 2.0)
@ -251,9 +238,9 @@ def _backtest_iron_condor(
put_spread_details['close'] = current_put_spread_price
if not call_spread_stopped_out:
if current_call_spread_price >= ((call_spread_strat.stop_loss_multiple + 1) * original_call_spread_price):
premium_received -= original_call_spread_price * (call_spread_strat.stop_loss_multiple + 1)
call_spread_details['close'] = original_call_spread_price * (call_spread_strat.stop_loss_multiple + 1) + 0.10
if current_call_spread_price >= ((call_spread_strategy.stop_loss_multiple + 1) * original_call_spread_price):
premium_received -= original_call_spread_price * (call_spread_strategy.stop_loss_multiple + 1)
call_spread_details['close'] = original_call_spread_price * (call_spread_strategy.stop_loss_multiple + 1) + 0.10
# Calculate exit slippage.
premium_received -= 0.10 # TODO: Make this configurable.
call_spread_stopped_out = True
@ -262,10 +249,10 @@ def _backtest_iron_condor(
break
if not put_spread_stopped_out:
if current_put_spread_price >= ((put_spread_strat.stop_loss_multiple + 1) * original_put_spread_price):
premium_received -= original_put_spread_price * (put_spread_strat.stop_loss_multiple + 1)
if current_put_spread_price >= ((put_spread_strategy.stop_loss_multiple + 1) * original_put_spread_price):
premium_received -= original_put_spread_price * (put_spread_strategy.stop_loss_multiple + 1)
premium_received -= 0.10 # TODO: Make this configurable.
put_spread_details['close'] = original_put_spread_price * (put_spread_strat.stop_loss_multiple + 1) + 0.10
put_spread_details['close'] = original_put_spread_price * (put_spread_strategy.stop_loss_multiple + 1) + 0.10
put_spread_stopped_out = True
exit_time = call_spread.name[-8:]
logging.info('Put Spread Stopped Out')
@ -283,19 +270,19 @@ def _backtest_iron_condor(
else:
current_call_spread_price = ((call_spread['ask_short_strike'] + call_spread['bid_short_strike']) / 2.0) - ((call_spread['ask_long_strike'] + call_spread['bid_long_strike']) / 2.0)
if call_spread_stopped_out:
current_call_spread_price = original_call_spread_price * (call_spread_strat.stop_loss_multiple + 1)
current_call_spread_price = original_call_spread_price * (call_spread_strategy.stop_loss_multiple + 1)
if put_spread['high_short_strike'] > put_spread['high_long_strike']:
current_put_spread_price = put_spread['high_short_strike'] - put_spread['high_long_strike']
else:
current_put_spread_price = ((put_spread['ask_short_strike'] + put_spread['bid_short_strike']) / 2.0) - ((put_spread['ask_long_strike'] + put_spread['bid_long_strike']) / 2.0)
if put_spread_stopped_out:
current_put_spread_price = original_put_spread_price * (put_spread_strat.stop_loss_multiple + 1)
current_put_spread_price = original_put_spread_price * (put_spread_strategy.stop_loss_multiple + 1)
if call_spread['high_short_strike'] > call_spread['high_long_strike']:
current_call_spread_price = call_spread['high_short_strike'] - call_spread['high_long_strike']
else:
current_call_spread_price = ((call_spread['ask_short_strike'] + call_spread['bid_short_strike']) / 2.0) - ((call_spread['ask_long_strike'] + call_spread['bid_long_strike']) / 2.0)
current_profit = (original_call_spread_price - current_call_spread_price) + (original_put_spread_price - current_put_spread_price)
current_profit_dollars = current_profit * call_spread_strat.number_of_contracts * 100
current_profit_dollars = current_profit * call_spread_strategy.number_of_contracts * 100
if current_profit_dollars > max_profit:
max_profit = current_profit_dollars
if current_profit_dollars < max_drawdown:
@ -308,7 +295,7 @@ def _backtest_iron_condor(
if not put_spread_stopped_out and current_put_spread_price > 0.05:
premium_received -= current_put_spread_price
number_of_contracts = call_spread_strat.number_of_contracts
number_of_contracts = call_spread_strategy.number_of_contracts
stop_out_fees = 0.0 # It costs money to get stopped out.
if call_spread_stopped_out:
stop_out_fees += (2 * FEES_PER_CONTRACT * number_of_contracts)
@ -341,9 +328,9 @@ def _backtest_iron_condor(
return result
def backtest_iron_condor(
strategy_name: str,
call_spread_strat: OptionStrat,
put_spread_strat: OptionStrat,
backtest_name: str,
call_spread_strategy: OptionSpreadStrategy,
put_spread_strategy: OptionSpreadStrategy,
start_date: datetime,
end_date: datetime
) -> pd.DataFrame:
@ -378,7 +365,7 @@ def backtest_iron_condor(
logging.info('Processing File: %s', historical_data_file)
historical_option_data = pd.read_csv(historical_data_file)
backtest_result = _backtest_iron_condor(historical_option_data, call_spread_strat, put_spread_strat)
backtest_result = _backtest_iron_condor(historical_option_data, call_spread_strategy, put_spread_strategy)
total_premium_received += backtest_result.trade_pnl
backtest_result.profit = total_premium_received
backtest_results.append(backtest_result)
@ -400,7 +387,7 @@ def backtest_iron_condor(
backtest_results = pd.DataFrame([{
'Date': result.date,
'Symbol': 'SPX',
'Strategy': strategy_name,
'Strategy': backtest_name,
'Entry Time': result.entry_time,
'Exit Time': result.exit_time,
'Spreads': result.spreads,
@ -409,44 +396,3 @@ def backtest_iron_condor(
} for result in backtest_results])
return backtest_results
def create_strategies(entry_time, number_of_contracts=1):
call_spread_strat = OptionStrat(
delta_upper_bound=0.11,
delta_lower_bound=0.10,
credit_target=1.50,
max_loss=5000,
number_of_contracts=number_of_contracts,
option_type=OptionType.CALL,
spread_width=50,
stop_loss_multiple=1.00,
stop_loss_percent=1.0,
trade_entry_time=entry_time
)
put_spread_strat = OptionStrat(
delta_upper_bound=0.11,
delta_lower_bound=0.10,
credit_target=1.50,
max_loss=5000,
number_of_contracts=number_of_contracts,
option_type=OptionType.PUT,
spread_width=50,
stop_loss_multiple=1.00,
stop_loss_percent=1.0,
trade_entry_time=entry_time
)
return call_spread_strat, put_spread_strat
if __name__ == '__main__':
start_date = datetime(2024, 1, 12)
end_date = datetime.now()
call_spread_strat, put_spread_strat = create_strategies(entry_time = '10:05:00')
backtest_result = backtest_iron_condor(
f'Iron Condor @ {call_spread_strat.trade_entry_time}',
call_spread_strat,
put_spread_strat,
start_date,
end_date
)
print(backtest_result)
plot(backtest_result, title = 'Iron Condor Backtest Results')

View File

@ -0,0 +1,7 @@
from dataclasses import dataclass
from .option_spread_strategy import OptionSpreadStrategy
@dataclass
class CreditTargetStrategy(OptionSpreadStrategy):
credit_target: float

View File

@ -0,0 +1,9 @@
from dataclasses import dataclass
from .option_spread_strategy import OptionSpreadStrategy
@dataclass
class DeltaTargetStrategy(OptionSpreadStrategy):
# TODO: Just search closest delta instead.
delta_upper_bound: float
delta_lower_bound: float

View File

@ -0,0 +1,11 @@
from dataclasses import dataclass
from .option_type import OptionType
@dataclass
class OptionSpreadStrategy:
option_type: OptionType
number_of_contracts: int
spread_width: int
stop_loss_multiple: float
trade_entry_time: str

View File

@ -0,0 +1,5 @@
from enum import Enum
class OptionType(Enum):
PUT = 'P'
CALL = 'C'