wassname/options_backtester
1
0
Fork 0
mirror of https://github.com/wassname/options_backtester.git synced 2026-07-25 13:20:23 +08:00
Code Issues Packages Projects Releases Wiki Activity

Replaced straddle class with the more general strangle and moved 'date' column to 'totals' index

Browse Source
This commit is contained in:
Javier Rodríguez Chatruc
2020-01-07 10:58:42 -03:00
parent bd268cc23c
commit b7576e87c7
5 changed files with 79 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backtester/backtester.py
+32 -13
View File
@@ -1,6 +1,8 @@
import pandas as pd
import numpy as np
import pyprind
import seaborn as sns
import matplotlib.pyplot as plt
from .strategy import Strategy
from .datahandler import HistoricalOptionsData
@@ -49,17 +51,17 @@ class Backtest:
index = pd.MultiIndex.from_product(
[[l.name for l in self._strategy.legs],
['contract', 'underlying', 'expiration', 'type', 'strike', 'cost', 'date', 'order']])
index_totals = pd.MultiIndex.from_product([['totals'], ['cost', 'qty']])
['contract', 'underlying', 'expiration', 'type', 'strike', 'cost', 'order']])
index_totals = pd.MultiIndex.from_product([['totals'], ['cost', 'qty', 'date']])
self.inventory = pd.DataFrame(columns=index.append(index_totals))
self.trade_log = pd.DataFrame()
data_iterator = self._data.iter_months() if monthly else self._data.iter_dates()
bar = pyprind.ProgBar(data_iterator.ngroups, bar_char='█')
for _date, options in data_iterator:
entry_signals = self._strategy.filter_entries(options, self.inventory)
exit_signals = self._strategy.filter_exits(options, self.inventory)
for date, options in data_iterator:
entry_signals = self._strategy.filter_entries(options, self.inventory, date)
exit_signals = self._strategy.filter_exits(options, self.inventory, date)
self._execute_exit(exit_signals)
self._execute_entry(entry_signals)
@@ -97,23 +99,28 @@ class Backtest:
return entry_signals, 0
def summary(self):
"""Returns a table with summary statistics about the trade log"""
df = self.trade_log
df.loc[:, ('totals', 'capital')] = (-df['totals']['cost']).cumsum() + self.initial_capital
df.loc[:, ('totals', 'return')] = (df['totals']['capital'].pct_change() * 100)
df.loc[:,
('totals',
'capital')] = (-df['totals']['cost'] * df['totals']['qty']).cumsum() + self._strategy.initial_capital
daily_df = df.groupby(('totals', 'date'))
daily_capital = daily_df.apply(lambda row: row['totals']['capital'].tail(1))
daily_returns = daily_capital.pct_change() * 100
entries_mask = df.apply(lambda row: row['leg_1']['order'][2] == 'O', axis=1)
entries = df.loc[entries_mask]
exits = df.loc[~entries_mask]
costs = np.array([])
returns = np.array([])
for contract in entries['leg_1']['contract']:
entry = entries.loc[entries['leg_1']['contract'] == contract]
exit_ = exits.loc[exits['leg_1']['contract'] == contract]
try:
# Here we assume we are entering only once per contract (i.e both entry and exit_ have only one row)
costs = np.append(costs, entry['totals']['cost'].values[0] + exit_['totals']['cost'].values[0])
returns = np.append(returns, exit_['totals']['return'])
costs = np.append(costs, (entry['totals']['cost'] * entry['totals']['qty']).values[0] +
(exit_['totals']['cost'] * exit_['totals']['qty']).values[0])
except IndexError:
continue
@@ -132,9 +139,8 @@ class Backtest:
win_pct = win_number / total_trades
largest_loss = np.max(costs)
avg_profit = np.sum(-costs) / len(costs)
profit_loss = returns
avg_pl = np.mean(profit_loss)
total_pl = np.sum(profit_loss)
avg_pl = np.mean(daily_returns)
total_pl = (df['totals']['capital'].iloc[-1] / self._strategy.initial_capital) * 100
data = [
total_trades, win_number, loss_number, win_pct, largest_loss, profit_factor, avg_profit, avg_pl, total_pl
@@ -145,6 +151,19 @@ class Backtest:
]
strat = ['Strategy']
summary = pd.DataFrame(data, stats, strat)
daily_returns = daily_returns[1:].reset_index(level=1, drop=True)
daily_returns_df = pd.DataFrame(data=daily_returns.groupby(daily_returns.index.year).apply(list).array,
index=daily_returns.index.year.unique(),
columns=[
'January', 'February', 'March', 'April', 'May', 'June', 'July', 'August',
'September', 'October', 'November', 'December'
])
sns.heatmap(daily_returns_df, linewidth=0.5, annot=True, fmt='f', cmap='YlGnBu', cbar=False)
plt.title('Monthly returns heatmap (in percentage)')
plt.show()
return summary
def __repr__(self):
backtester/strategy/__init__.py
+1
View File
@@ -1,3 +1,4 @@
from .strategy import Strategy, Condition
from .strategy_leg import StrategyLeg
from .straddle import Straddle
from .strangle import Strangle
backtester/strategy/straddle.py
-41
View File
@@ -1,41 +0,0 @@
from .strategy_leg import StrategyLeg
from .strategy import Strategy
from backtester.option import Direction, Type
class Straddle(Strategy):
def __init__(self,
schema,
name,
underlying,
dte_entry_range,
dte_exit,
atm_pct=(0.05, 0.05),
exit_thresholds=(float('inf'), float('inf')),
qty=1,
shares_per_contract=100):
assert (name.lower() == 'short' or name.lower() == 'long')
super().__init__(schema, qty, shares_per_contract)
direction = Direction.SELL if name.lower() == 'short' else Direction.BUY
leg1 = StrategyLeg(
"leg_1",
schema,
option_type=Type.CALL,
direction=direction,
)
leg1.entry_filter = (schema.underlying == underlying) & (schema.dte >= dte_entry_range[0]) & (
schema.dte <= dte_entry_range[1]) & (schema.strike >= schema.underlying_last *
(1 - atm_pct[0])) & (schema.strike <= schema.underlying_last *
(1 + atm_pct[1]))
leg1.exit_filter = (schema.dte <= dte_exit)
leg2 = StrategyLeg("leg_2", schema, option_type=Type.PUT, direction=direction)
leg2.entry_filter = (schema.underlying == underlying) & (schema.dte >= dte_entry_range[0]) & (
schema.dte <= dte_entry_range[1]) & (schema.strike >= schema.underlying_last *
(1 - atm_pct[0])) & (schema.strike <= schema.underlying_last *
(1 + atm_pct[1]))
leg2.exit_filter = (schema.dte <= dte_exit)
self.add_legs([leg1, leg2])
self.add_exit_thresholds(exit_thresholds[0], exit_thresholds[1])
backtester/strategy/strangle.py
+36 -71
View File
@@ -1,80 +1,45 @@
import pandas as pd
from .strategy_leg import StrategyLeg
from .strategy import Strategy
from backtester.option import Direction, Type
class Strangle:
class Strangle(Strategy):
def __init__(self,
schema,
name,
underlying,
strike,
dte,
strike_diff,
shares_per_contract=100,
capital=1000000.0):
self.underlying = underlying
self.strike = strike
self.dte = dte
self.strike_diff = strike_diff
self.inventory = set()
self.shares_per_contract = shares_per_contract
self.capital = capital
dte_entry_range,
dte_exit,
otm_pct=0,
pct_tolerance=1,
exit_thresholds=(float('inf'), float('inf')),
shares_per_contract=100):
assert (name.lower() == 'short' or name.lower() == 'long')
super().__init__(schema, shares_per_contract)
direction = Direction.SELL if name.lower() == 'short' else Direction.BUY
def execute_entry(self, date, group):
calls = group.loc[(group.type == 'call')
& (group.strike >= self.strike[0]) &
(group.strike <= self.strike[1]) &
(group.dte >= self.dte[0])
& (group.dte <= self.dte[1])]
puts = group.loc[group.type == 'put']
merge = calls.merge(puts, on=['dte'], suffixes=('_call', '_put'))
merge['ask_sum'] = merge['ask_call'] + merge['ask_put']
merge['strike_diff'] = abs(merge['strike_call'] - merge['strike_put'])
merge_strangle = merge.loc[merge['strike_diff'] <= self.strike_diff]
if merge_strangle.empty:
return
entry_index = merge_strangle['ask_sum'].idxmin()
entry = merge_strangle.loc[entry_index]
cost = sum([entry['ask_sum'] * self.shares_per_contract])
if cost <= self.capital:
self.capital -= cost
self.inventory.add((entry.optionroot_call, entry.dte))
self.inventory.add((entry.optionroot_put, entry.dte))
self._update_trade_log(date, entry.optionroot_call,
entry.type_call,
-entry.ask_call * self.shares_per_contract)
self._update_trade_log(date, entry.optionroot_put, entry.type_put,
-entry.ask_put * self.shares_per_contract)
leg1 = StrategyLeg(
"leg_1",
schema,
option_type=Type.CALL,
direction=direction,
)
def execute_exits(self, inventory, date, group):
exits = []
remove_set = set()
for entry in inventory:
exit = group.loc[(group.optionroot == entry[0]) & (group.dte == 1)]
if not exit.empty:
exits.append(exit)
remove_set.add(entry)
for exit in exits:
profit = exit.bid.values[0] * self.shares_per_contract
contract = exit.optionroot.values[0]
type_ = exit.type.values[0]
self.capital += profit
self._update_trade_log(date, contract, type_, profit)
self.inventory.difference_update(remove_set)
otm_lower_bound = (otm_pct - pct_tolerance) / 100
otm_upper_bound = (otm_pct + pct_tolerance) / 100
def run(self, data):
self.trade_log = pd.DataFrame(
columns=["date", "contract", "type", "profit", "capital"])
leg1.entry_filter = (schema.underlying == underlying) & (schema.dte >= dte_entry_range[0]) & (
schema.dte <= dte_entry_range[1]) & (schema.strike >= schema.underlying_last *
(1 + otm_lower_bound)) & (schema.strike <= schema.underlying_last *
(1 + otm_upper_bound))
leg1.exit_filter = (schema.dte <= dte_exit)
for date, group in self._iter_dates(data):
self.execute_entry(date, group)
self.execute_exits(self.inventory, date, group)
leg2 = StrategyLeg("leg_2", schema, option_type=Type.PUT, direction=direction)
leg2.entry_filter = (schema.underlying == underlying) & (schema.dte >= dte_entry_range[0]) & (
schema.dte <= dte_entry_range[1]) & (schema.strike <= schema.underlying_last *
(1 - otm_lower_bound)) & (schema.strike >= schema.underlying_last *
(1 - otm_upper_bound))
leg2.exit_filter = (schema.dte <= dte_exit)
return self.trade_log
def _update_trade_log(self, date, contract, type_, profit):
"""Adds entry for the given order to `self.trade_log`."""
self.trade_log.loc[len(
self.trade_log)] = [date, contract, type_, profit, self.capital]
def _iter_dates(self, data):
"""Returns `pd.DataFrameGroupBy` with the given underlying and with contracts grouped by date"""
df = data._data.loc[data._data.underlying == self.underlying]
return df.groupby(data.schema["date"])
self.add_legs([leg1, leg2])
self.add_exit_thresholds(exit_thresholds[0], exit_thresholds[1])
backtester/strategy/strategy.py
+10 -11
View File
@@ -18,10 +18,9 @@ class Strategy:
Takes in a number of `StrategyLeg`'s (option contracts), and filters that determine
entry and exit conditions.
"""
def __init__(self, schema, qty=1, shares_per_contract=100, initial_capital=1_000_000):
def __init__(self, schema, shares_per_contract=100, initial_capital=1_000_000):
assert isinstance(schema, Schema)
self.schema = schema
self.qty = qty
self._shares_per_contract = shares_per_contract
self.initial_capital = initial_capital
self.legs = []
@@ -74,7 +73,7 @@ class Strategy:
assert loss_pct >= 0
self.exit_thresholds = (profit_pct, loss_pct)
def filter_entries(self, options, inventory):
def filter_entries(self, options, inventory, date):
"""Returns the entry signals chosen by the strategy for the given
(daily) options.
@@ -89,9 +88,9 @@ class Strategy:
inventory_contracts = pd.concat([inventory[leg.name]['contract'] for leg in self.legs])
subset_options = options[~options[self.schema['contract']].isin(inventory_contracts)]
return self._filter_legs(subset_options, Signal.ENTRY)
return self._filter_legs(subset_options, Signal.ENTRY, date)
def filter_exits(self, options, inventory):
def filter_exits(self, options, inventory, date):
"""Returns the exit signals chosen by the strategy for the given
(daily) options.
@@ -123,7 +122,8 @@ class Strategy:
leg_candidates[i].columns])
qtys = inventory['totals']['qty']
totals = pd.DataFrame.from_dict({"cost": total_costs, "qty": qtys})
dates = [date] * len(inventory)
totals = pd.DataFrame.from_dict({"cost": total_costs, "qty": qtys, "date": dates})
totals.columns = pd.MultiIndex.from_product([["totals"], totals.columns])
leg_candidates.append(totals)
filter_mask = reduce(lambda x, y: x | y, filter_mask)
@@ -134,7 +134,7 @@ class Strategy:
return (exits, exits_mask, total_costs)
def _filter_legs(self, options, signal):
def _filter_legs(self, options, signal, date):
"""Returns a hierarchically indexed `pd.DataFrame` containing signals for each
leg in the strategy.
@@ -171,7 +171,7 @@ class Strategy:
dfs.append(subset_df.reset_index(drop=True))
return self._apply_conditions(dfs)
return self._apply_conditions(dfs, date)
def _signal_fields(self, cost_field):
fields = {
@@ -181,13 +181,12 @@ class Strategy:
self.schema['type']: 'type',
self.schema['strike']: 'strike',
self.schema[cost_field]: 'cost',
self.schema['date']: 'date',
'order': 'order'
}
return fields
def _apply_conditions(self, dfs):
def _apply_conditions(self, dfs, date):
"""Applies conditions on the specified legs."""
for condition in self.conditions:
@@ -211,7 +210,7 @@ class Strategy:
qty = np.floor(self.initial_capital / cost)
qty = np.abs(qty)
# qty = qty.astype(int)
totals = pd.DataFrame.from_dict({"cost": cost, "qty": qty})
totals = pd.DataFrame.from_dict({"cost": cost, "qty": qty, "date": date})
totals.columns = pd.MultiIndex.from_product([["totals"], totals.columns])
for i in range(len(dfs)):