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

Working on Strategy implementation. Added Historical Options Data handler.

Browse Source
This commit is contained in:
Juan Pablo Amoroso
2019-05-30 15:36:53 -03:00
parent b1e6589ee6
commit 8a26e32d0f
9 changed files with 104 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backtester/datahandler/balanced_datahandler.py
-39
View File
@@ -1,39 +0,0 @@
import pandas as pd
from .datahandler import DataHandler
from ..event import MarketEvent
class BalancedDataHandler(DataHandler):
"""Handler for balanced data set"""
def __init__(self, data_path, events):
data = pd.read_csv(data_path, parse_dates=["date"])
# We will assume bid and ask prices = close
data["bid"] = data["close"]
data["ask"] = data["close"]
self._data_generator = self._get_data_generator(data)
self.events = events
self.continue_backtest = True
def get_latest_bars(self, symbol, N=1):
"""Returns the latest `N` bars for `symbol` if there are at least N
rows, otherwise returns the all data.
Returns empty dataframe if `symbol` is not in self.data.
"""
return self._current_bar[self._current_bar["symbol"] == symbol].iloc[0]
def update_bars(self):
"""Add new data bar to self.data"""
try:
self.current_date, self._current_bar = next(self._data_generator)
self.events.put(MarketEvent())
except StopIteration:
self.continue_backtest = False
def _get_data_generator(self, data):
"""Returns generator that yields daily data bars"""
grouped = data.groupby("date")
for date, bars in grouped:
yield date, bars
backtester/datahandler/datahandler.py
-21
View File
@@ -1,21 +0,0 @@
from abc import ABCMeta, abstractmethod
class DataHandler(metaclass=ABCMeta):
"""Interface for the different data handlers"""
@abstractmethod
def get_latest_bars(self, symbol, N=1):
"""
Returns the last N bars from the latest_symbol list,
or fewer if less bars are available.
"""
raise NotImplementedError("Should implement get_latest_bars()")
@abstractmethod
def update_bars(self):
"""
Pushes the latest bar to the latest symbol structure
for all symbols in the symbol list.
"""
raise NotImplementedError("Should implement update_bars()")
backtester/datahandler/historic_datahandler.py
-35
View File
@@ -1,35 +0,0 @@
import pandas as pd
from .datahandler import DataHandler
from ..event import MarketEvent
class HistoricDataHandler(DataHandler):
"""Handler for Historical Option Data"""
def __init__(self, data_path, events):
self._data = pd.read_csv(
data_path, parse_dates=["quotedate",
"expiration"]).sort_values(by="date")
columns = {"quotedate": "date", "optionroot": "symbol"}
self._data.rename(columns=columns, inplace=True)
self._data_index = 0
self.events = events
self.continue_backtest = True
def get_latest_bars(self, symbol, N=1):
"""Returns the latest `N` bars for `symbol` if there are at least N
rows, otherwise returns the all data.
Returns empty dataframe if `symbol` is not in self._data.
"""
return self._data[(self._data["symbol"] == symbol)
& (self._data["date"] <= self.current_date)][-N:]
def update_bars(self):
"""Add new data bar to self.data"""
if self._data_index < len(self._data):
self.current_date = self._data["date"][self._data_index]
self.events.put(MarketEvent())
self._data_index += 1
else:
self.continue_backtest = False
backtester/datahandler/historical_options_data.py
+21 -5
View File
@@ -9,22 +9,38 @@ class HistoricalOptionsData:
if schema:
assert isinstance(schema, Schema)
else:
schema = Schema.canonical()
schema.update({"contract": "optionroot", "date": "quotedate"})
self.schema = schema
self.schema = HistoricalOptionsData.default_schema()
self._data = pd.read_hdf(file, **params)
columns = self._data.columns
assert all((col in columns for col in schema))
assert all((col in columns for _key, col in self.schema))
self._data["dte"] = (self._data["expiration"] -
self._data["quotedate"]).dt.days
self.schema.update({"dte": "dte"})
def __getitem__(self, item):
return self._data[item]
key = self.schema[item].mapping
return self._data[key]
def __setitem__(self, item, value):
self._data[item] = value
def __repr__(self):
return self._data.__repr__()
def default_schema():
"""Returns default schema for Historical Options Data"""
schema = Schema.canonical()
schema.update({
"contract": "optionroot",
"date": "quotedate",
"last": "last",
"open_interest": "openinterest",
"impliedvol": "impliedvol",
"delta": "delta",
"gamma": "gamma",
"theta": "theta",
"vega": "vega"
})
return schema
backtester/datahandler/schema.py
+9 -2
View File
@@ -3,7 +3,7 @@ class Schema:
columns = [
"underlying", "underlying_last", "date", "contract", "type",
"expiration", "strike", "bid", "ask"
"expiration", "strike", "bid", "ask", "volume", "open_interest"
]
def canonical():
@@ -26,7 +26,7 @@ class Schema:
return Field(key, self._mappings[key])
def __iter__(self):
return iter(self._mappings.values())
return iter(self._mappings.items())
def __repr__(self):
return "Schema({})".format(
@@ -77,17 +77,24 @@ class Filter:
self.query = query
def __and__(self, other):
"""Returns logical *and* between `self` and `other`"""
assert isinstance(other, Filter)
new_query = "({}) & ({})".format(self.query, other.query)
return Filter(query=new_query)
def __or__(self, other):
"""Returns logical *or* between `self` and `other`"""
assert isinstance(other, Filter)
new_query = "(({}) | ({}))".format(self.query, other.query)
return Filter(query=new_query)
def __invert__(self):
"""Negates filter"""
return Filter("!({})".format(self.query))
def __call__(self, data):
"""Returns filtered dataframe"""
return data.query(self.query)
def __repr__(self):
return "Filter(query='{}')".format(self.query)
backtester/datahandler/spx_datahandler.py
-33
View File
@@ -1,33 +0,0 @@
import pandas as pd
from .datahandler import DataHandler
from ..event import MarketEvent
class SPXDataHandler(DataHandler):
"""Handler for SPX test data"""
def __init__(self, data_path, events):
self._data = pd.read_csv(
data_path, parse_dates=["date"]).sort_values(by="date")
self._data.rename(columns={"price": "ask"}, inplace=True)
self._data["bid"] = self._data["ask"]
self._data_index = 0
self.events = events
self.continue_backtest = True
def get_latest_bars(self, symbol, N=1):
"""Returns the latest `N` bars for `symbol` if there are at least N
rows, otherwise returns the all data.
Returns empty dataframe if `symbol` is not in self.data.
"""
return self._data[self._data["date"] <= self.current_date][-N:]
def update_bars(self):
"""Add new data bar to self.data"""
if self._data_index < len(self._data):
self.current_date = self._data["date"][self._data_index]
self.events.put(MarketEvent())
self._data_index += 1
else:
self.continue_backtest = False
backtester/option.py
+31
View File
@@ -0,0 +1,31 @@
from enum import Enum
class OptionContract:
"""Option contract data class"""
Type = Enum("Type", {"CALL": "call", "PUT": "put"})
Direction = Enum("Direction", "BUY SELL")
# Orders:
# BTO: Buy to Open
# BTC: Buy to Close
# STO: Sell to Open
# STC: Sell to Close
Order = Enum("Order", "BTO BTC STO STC")
def __init__(self,
option_type=Type.CALL,
direction=Direction.BUY,
order=Order.BTO):
assert isinstance(option_type, OptionContract.Type)
assert isinstance(direction, OptionContract.Direction)
assert isinstance(order, OptionContract.Order)
self._store = {}
self._store["type"] = option_type
self._store["direction"] = direction
self._store["order"] = order
def __repr__(self):
return "Option({})".format(str(self._store))
backtester/strategy/__init__.py
-2
View File
@@ -1,3 +1 @@
from .strategy import Strategy
from .benchmark import Benchmark
from .balanced import Balanced
backtester/strategy/strategy.py
+43 -7
View File
@@ -1,11 +1,47 @@
from abc import ABCMeta, abstractmethod
from ..option import OptionContract
from ..datahandler import Filter
class Strategy(metaclass=ABCMeta):
"""Interface for the different investing strategies"""
class Strategy:
"""Options strategy class.
Takes in a number of `legs` (option contracts), and filters that determine
entry and exit conditions.
"""
@abstractmethod
def generate_signals(self, event):
"""Provides the mechanisms to calculate the list of signals.
def __init__(self, data, entry_filter, exit_filter, legs=[]):
assert all((isinstance(leg, OptionContract) for leg in legs))
assert isinstance(entry_filter, Filter)
assert isinstance(exit_filter, Filter)
self.data = data
self.entry = entry_filter
self.exit = exit_filter
self.legs = legs
def add_leg(self, leg):
"""Adds leg to the strategy"""
self.legs.append(leg)
return self
def remove_leg(self, leg_number):
"""Removes leg to the strategy"""
self.legs.pop(leg_number)
return self
def run(self, data):
"""Returns a dataframe of trades executed as a result of
runnning the strategy on the data.
"""
raise NotImplementedError("Strategy must implement generate_signals()")
entry_query = self.entry(self._data)
exit_query = self.exit(self._data)
entry_df = data.query(entry_query)
exit_df = data.query(exit_query)
return entry_df.merge(exit_df,
on="optionroot",
suffixes=("_entry", "_exit"))
def __repr__(self):
return "Strategy(entry_filter={}, exit_filter={}, legs={})".format(
self.entry, self.exit, self.legs)