mirror of
https://github.com/wassname/options_backtester.git
synced 2026-06-27 18:05:27 +08:00
Added first draft of backtester
This commit is contained in:
Juan Pablo Amoroso
@@ -0,0 +1 @@
|
||||
from .backtester import *
|
||||
@@ -0,0 +1,15 @@
|
||||
import argparse
|
||||
import os
|
||||
import logging
|
||||
from .backtester import run
|
||||
from .utils import get_data_dir
|
||||
|
||||
parser = argparse.ArgumentParser(prog="backtester.py")
|
||||
parser.add_argument(
|
||||
"-t", "--symbols", nargs="+", help="Symbols to fetch", required=True)
|
||||
parser.add_argument("-s", "--scraper", choices=["cboe"])
|
||||
args = parser.parse_args()
|
||||
|
||||
data_dir = get_data_dir()
|
||||
spx_data = os.path.join(data_dir, "SPX_2008-2018.csv")
|
||||
run(spx_data)
|
||||
@@ -0,0 +1,45 @@
|
||||
"""Event based backtester"""
|
||||
|
||||
from queue import Queue
|
||||
from .datahandler import BalancedDataHandler
|
||||
from .strategy import Balanced
|
||||
from .portfolio import BalancedPortfolio
|
||||
|
||||
|
||||
def run(data_path,
|
||||
data_handler=BalancedDataHandler,
|
||||
port_class=BalancedPortfolio,
|
||||
strat_class=Balanced,
|
||||
**strat_args):
|
||||
events = Queue()
|
||||
bars = data_handler(data_path, events)
|
||||
|
||||
weights = {
|
||||
"VOO": 0.3,
|
||||
"GLD": 0.1,
|
||||
"VNQ": 0.05,
|
||||
"VNQI": 0.05,
|
||||
"TLT": 0.2,
|
||||
"TIP": 0.1,
|
||||
"BNDX": 0.1,
|
||||
"RJI": 0.1
|
||||
}
|
||||
port = port_class(bars, events, weights=weights)
|
||||
strat = strat_class(bars, events, **strat_args)
|
||||
|
||||
while True:
|
||||
bars.update_bars()
|
||||
if not bars.continue_backtest:
|
||||
break
|
||||
|
||||
while True:
|
||||
if events.empty():
|
||||
break
|
||||
event = events.get()
|
||||
if event.type == "MARKET":
|
||||
strat.generate_signals(event)
|
||||
port.update_timeindex(event)
|
||||
elif event.type == "SIGNAL":
|
||||
port.update_signal(event)
|
||||
|
||||
return port
|
||||
@@ -0,0 +1,4 @@
|
||||
from .datahandler import DataHandler
|
||||
from .historic_datahandler import HistoricDataHandler
|
||||
from .spx_datahandler import SPXDataHandler
|
||||
from .balanced_datahandler import BalancedDataHandler
|
||||
@@ -0,0 +1,39 @@
|
||||
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
|
||||
@@ -0,0 +1,21 @@
|
||||
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()")
|
||||
@@ -0,0 +1,35 @@
|
||||
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
|
||||
@@ -0,0 +1,33 @@
|
||||
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
|
||||
@@ -0,0 +1,34 @@
|
||||
class Event():
|
||||
"""
|
||||
Event is base class providing an interface for all subsequent
|
||||
(inherited) events, that will trigger further events in the
|
||||
trading infrastructure.
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class MarketEvent(Event):
|
||||
"""
|
||||
Handles the event of receiving a new market update with
|
||||
corresponding bars.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.type = "MARKET"
|
||||
|
||||
|
||||
class SignalEvent(Event):
|
||||
"""
|
||||
Handles the event of receiving a signal form the Strategy
|
||||
object.
|
||||
Portfolio object processes buy/sell orders.
|
||||
"""
|
||||
|
||||
def __init__(self, symbol, direction, strength):
|
||||
"""symbol: ticker symbol
|
||||
direction: BUY | SELL
|
||||
strength: (%Win chance, Win/Loss ratio)"""
|
||||
self.type = "SIGNAL"
|
||||
self.symbol = symbol
|
||||
self.direction = direction
|
||||
self.strength = strength
|
||||
@@ -0,0 +1,4 @@
|
||||
from .portfolio import Portfolio
|
||||
from .kellyportfolio import KellyPortfolio
|
||||
from .simpleportfolio import SimplePortfolio
|
||||
from .balancedportfolio import BalancedPortfolio
|
||||
@@ -0,0 +1,17 @@
|
||||
from .portfolio import Portfolio
|
||||
|
||||
|
||||
class BalancedPortfolio(Portfolio):
|
||||
"""Buys and holds a basket of securities, and allocates them
|
||||
according to given weights.
|
||||
"""
|
||||
|
||||
def __init__(self, *args, weights={}):
|
||||
self.weights = weights
|
||||
super().__init__(*args)
|
||||
|
||||
def _get_allocation(self, signal, price):
|
||||
"""Allocates capital in porportion to given weight"""
|
||||
weight = self.weights.get(signal.symbol, 0)
|
||||
cash_proportion = self.initial_capital * weight
|
||||
return cash_proportion / price
|
||||
@@ -0,0 +1,16 @@
|
||||
import math
|
||||
from .portfolio import Portfolio
|
||||
|
||||
|
||||
class KellyPortfolio(Portfolio):
|
||||
"""Allocates signals using Kelly's criterion"""
|
||||
|
||||
def __init__(self, *args):
|
||||
super().__init__(*args)
|
||||
|
||||
def _get_allocation(self, strength, price):
|
||||
"""Calculates allocation using Kelly's criterion"""
|
||||
(win_percent, win_loss_ratio) = strength
|
||||
kelly = max(0, win_percent - (1 - win_percent) / win_loss_ratio)
|
||||
total_allocation = self.current_position["Cash"] * kelly
|
||||
return math.floor(total_allocation / price)
|
||||
@@ -0,0 +1,88 @@
|
||||
from abc import ABCMeta, abstractmethod
|
||||
import pandas as pd
|
||||
|
||||
|
||||
class Portfolio(metaclass=ABCMeta):
|
||||
"""Processes signals from the Strategy object"""
|
||||
|
||||
@abstractmethod
|
||||
def __init__(self, data_handler, events, capital=1000000):
|
||||
self.data_handler = data_handler
|
||||
self.events = events
|
||||
self.initial_capital = capital
|
||||
self.current_position = {"Cash": self.initial_capital}
|
||||
self.all_positions = {}
|
||||
self.current_balance = {"Cash": self.initial_capital}
|
||||
self.all_balances = {}
|
||||
|
||||
@abstractmethod
|
||||
def _get_allocation(self, strength, price):
|
||||
"""Calculates symbol allocation"""
|
||||
raise NotImplementedError("Portfolio must implement _get_allocation()")
|
||||
|
||||
def update_signal(self, signal):
|
||||
"""Processes signal event and updates the current position"""
|
||||
date = self.data_handler.current_date
|
||||
if date not in self.all_positions:
|
||||
self.all_positions[date] = self.current_position.copy()
|
||||
self.current_position = self.all_positions[date]
|
||||
|
||||
(price, direction) = self._get_price(signal)
|
||||
qty = self._get_allocation(signal, price)
|
||||
(current_amount, current_open_price) = self.current_position.get(
|
||||
signal.symbol, (0, 0))
|
||||
new_open_price = (current_open_price * current_amount +
|
||||
direction * price * qty) / (current_amount + qty)
|
||||
self.current_position[signal.symbol] = (
|
||||
current_amount + direction * qty, new_open_price)
|
||||
self.current_position["Cash"] -= direction * price * qty
|
||||
|
||||
def update_timeindex(self, event):
|
||||
"""Calculates new balance for the current timeindex.
|
||||
Appends current position to all_positions list."""
|
||||
date = self.data_handler.current_date
|
||||
self.all_balances[date] = self.current_balance.copy()
|
||||
self.current_balance = self.all_balances[date]
|
||||
self.current_balance["Total Exposure"] = 0
|
||||
|
||||
for symbol, values in self.current_position.items():
|
||||
if symbol == "Cash":
|
||||
self.current_balance["Cash"] = values
|
||||
continue
|
||||
|
||||
(amount, open_price) = values
|
||||
current_bar = self.data_handler.get_latest_bars(symbol)
|
||||
if amount < 0:
|
||||
price = current_bar["ask"]
|
||||
else:
|
||||
price = current_bar["bid"]
|
||||
market_value = amount * price
|
||||
self.current_balance[symbol + " Amount"] = amount
|
||||
self.current_balance[symbol + " Open"] = open_price
|
||||
self.current_balance[symbol + " Exposure"] = market_value
|
||||
self.current_balance["Total Exposure"] += market_value
|
||||
|
||||
self.all_positions[date] = self.current_position
|
||||
self.all_balances[date] = self.current_balance
|
||||
|
||||
def _get_price(self, signal):
|
||||
"""Returns price and direction for given symbol.
|
||||
Ask price if signal.type == BUY, bid price if signal.type == SELL.
|
||||
Also returns 1 or -1 for types BUY, SELL respectively"""
|
||||
current_bar = self.data_handler.get_latest_bars(signal.symbol)
|
||||
if signal.direction == "BUY":
|
||||
direction = 1
|
||||
price = current_bar["ask"]
|
||||
else:
|
||||
direction = -1
|
||||
price = current_bar["bid"]
|
||||
return (price, direction)
|
||||
|
||||
def create_report(self):
|
||||
"""Creates a pandas DataFrame from all_balances."""
|
||||
curve = pd.DataFrame(self.all_balances)
|
||||
curve = curve.transpose()
|
||||
curve["Total Portfolio"] = curve["Total Exposure"] + curve["Cash"]
|
||||
curve["Interval Change"] = curve["Total Portfolio"].pct_change()
|
||||
curve["% Price"] = (1.0 + curve["Interval Change"]).cumprod() - 1
|
||||
return curve
|
||||
@@ -0,0 +1,13 @@
|
||||
import math
|
||||
from .portfolio import Portfolio
|
||||
|
||||
|
||||
class SimplePortfolio(Portfolio):
|
||||
"""Allocates all capital to the first signal processed"""
|
||||
|
||||
def __init__(self, *args):
|
||||
super().__init__(*args)
|
||||
|
||||
def _get_allocation(self, strength, price):
|
||||
"""Allocates all capital to the given signal"""
|
||||
return math.floor(self.current_position["Cash"] / price)
|
||||
@@ -0,0 +1,80 @@
|
||||
"""Generates charts from a portfolio report"""
|
||||
|
||||
import altair as alt
|
||||
|
||||
|
||||
def returns_chart(report):
|
||||
# Time interval selector
|
||||
time_interval = alt.selection(type="interval", encodings=["x"])
|
||||
|
||||
# Area plot
|
||||
areas = alt.Chart().mark_area(opacity=0.6).encode(
|
||||
x=alt.X(
|
||||
"index:T",
|
||||
axis=alt.Axis(title="Date"),
|
||||
scale={"domain": time_interval.ref()}),
|
||||
y=alt.Y("% Price:Q", axis=alt.Axis(format="%")))
|
||||
|
||||
# Nearest point selector
|
||||
nearest = alt.selection(
|
||||
type="single",
|
||||
nearest=True,
|
||||
on="mouseover",
|
||||
fields=["index"],
|
||||
empty="none")
|
||||
|
||||
points = areas.mark_point().encode(
|
||||
opacity=alt.condition(nearest, alt.value(1), alt.value(0)))
|
||||
|
||||
# Transparent date selector
|
||||
selectors = alt.Chart().mark_point().encode(
|
||||
x="index:T",
|
||||
opacity=alt.value(0),
|
||||
).add_selection(nearest)
|
||||
|
||||
text = areas.mark_text(
|
||||
align="left", dx=5, dy=-5).encode(
|
||||
text=alt.condition(nearest, "% Price:Q", alt.value(" ")))
|
||||
|
||||
layered = alt.layer(
|
||||
areas,
|
||||
selectors,
|
||||
points,
|
||||
text,
|
||||
width=700,
|
||||
height=350,
|
||||
title="Returns over time")
|
||||
|
||||
lower = areas.properties(width=700, height=70).add_selection(time_interval)
|
||||
|
||||
chart = alt.vconcat(layered, lower, data=report.reset_index())
|
||||
|
||||
return chart
|
||||
|
||||
|
||||
def returns_histogram(report):
|
||||
bar = alt.Chart(report).mark_bar().encode(
|
||||
x=alt.X(
|
||||
"Interval Change:Q",
|
||||
bin=alt.BinParams(maxbins=100),
|
||||
axis=alt.Axis(format='%')),
|
||||
y="count():Q")
|
||||
return bar
|
||||
|
||||
|
||||
def monthly_returns_heatmap(report):
|
||||
monthly_returns = report.resample(
|
||||
"M")["Total Portfolio"].last().pct_change().reset_index()
|
||||
monthly_returns.columns = ["Date", "Monthly Returns"]
|
||||
|
||||
chart = alt.Chart(monthly_returns).mark_rect().encode(
|
||||
alt.X("year(Date):O", title="Year"),
|
||||
alt.Y("month(Date):O", title="Month"),
|
||||
alt.Color(
|
||||
"mean(Monthly Returns)",
|
||||
title="Return",
|
||||
scale=alt.Scale(scheme="redyellowgreen")),
|
||||
alt.Tooltip("mean(Monthly Returns)",
|
||||
format=".2f")).properties(title="Average Monthly Returns")
|
||||
|
||||
return chart
|
||||
@@ -0,0 +1,3 @@
|
||||
from .strategy import Strategy
|
||||
from .benchmark import Benchmark
|
||||
from .balanced import Balanced
|
||||
@@ -0,0 +1,29 @@
|
||||
from .strategy import Strategy
|
||||
from ..event import SignalEvent
|
||||
|
||||
|
||||
class Balanced(Strategy):
|
||||
"""Balanced portfolio strategy.
|
||||
Inspired by Ray Dalio's all weather portfolio.
|
||||
"""
|
||||
|
||||
def __init__(self,
|
||||
data_handler,
|
||||
events,
|
||||
symbols=[
|
||||
"VOO", "GLD", "VNQ", "VNQI", "TLT", "TIP", "BNDX", "EEM",
|
||||
"RJI"
|
||||
]):
|
||||
self.data_handler = data_handler
|
||||
self.symbols = symbols
|
||||
self.events = events
|
||||
self._bought = False
|
||||
|
||||
def generate_signals(self, event):
|
||||
if not self._bought:
|
||||
for symbol in self.symbols:
|
||||
buy_signal = SignalEvent(
|
||||
symbol=symbol, direction="BUY", strength=(1.0, 100))
|
||||
self.events.put(buy_signal)
|
||||
|
||||
self._bought = True
|
||||
@@ -0,0 +1,18 @@
|
||||
from .strategy import Strategy
|
||||
from ..event import SignalEvent
|
||||
|
||||
|
||||
class Benchmark(Strategy):
|
||||
"""Simple buy and hold SPX strategy"""
|
||||
|
||||
def __init__(self, data_handler, events):
|
||||
self.data_handler = data_handler
|
||||
self.events = events
|
||||
self._bought = False
|
||||
|
||||
def generate_signals(self, event):
|
||||
if not self._bought:
|
||||
buy_signal = SignalEvent(
|
||||
symbol="SPX", direction="BUY", strength=(1.0, 100))
|
||||
self.events.put(buy_signal)
|
||||
self._bought = True
|
||||
@@ -0,0 +1,11 @@
|
||||
from abc import ABCMeta, abstractmethod
|
||||
|
||||
|
||||
class Strategy(metaclass=ABCMeta):
|
||||
"""Interface for the different investing strategies"""
|
||||
|
||||
@abstractmethod
|
||||
def generate_signals(self, event):
|
||||
"""Provides the mechanisms to calculate the list of signals.
|
||||
"""
|
||||
raise NotImplementedError("Strategy must implement generate_signals()")
|
||||
@@ -0,0 +1,88 @@
|
||||
import os
|
||||
from datetime import date, timedelta
|
||||
import pandas as pd
|
||||
import pandas_datareader.data as web
|
||||
from backtester.utils import get_data_dir
|
||||
|
||||
|
||||
def create_test_data(data_dir, filename="SPX_2008-2018.csv"):
|
||||
"""Create test data set with 10 years of SPX"""
|
||||
|
||||
spx_dir = os.path.join(data_dir, "allspx")
|
||||
test_file = os.path.join(data_dir, filename)
|
||||
|
||||
with open(test_file, "w+") as f:
|
||||
f.write("date,price\n")
|
||||
|
||||
for year in range(2008, 2019):
|
||||
filename = "SPX_{}.csv".format(year)
|
||||
year_df = pd.read_csv(os.path.join(spx_dir, filename))
|
||||
grouped = year_df.groupby("quotedate").first()
|
||||
grouped.to_csv(
|
||||
test_file, mode="a", columns=["underlying_last"], header=False)
|
||||
|
||||
|
||||
def create_synthetic_data(data_dir, filename="synthetic_data.csv"):
|
||||
"""Create an synthetic data set with known statistics.
|
||||
Price goes from 1 to 2000.
|
||||
Mean = 1000.5
|
||||
% Price = 1999"""
|
||||
|
||||
synth_file = os.path.join(data_dir, filename)
|
||||
|
||||
day = date(1970, 1, 1)
|
||||
with open(synth_file, "w+") as f:
|
||||
f.write("date,price\n")
|
||||
for i in range(1, 2001):
|
||||
line = "{},{}\n".format(day.strftime("%m/%d/%Y"), i)
|
||||
f.write(line)
|
||||
day += timedelta(days=1)
|
||||
|
||||
|
||||
def fetch_balanced_data(data_dir,
|
||||
filename="balanced_2015.csv",
|
||||
start=None,
|
||||
end=None):
|
||||
"""Downloads daily data from `start` til `end` from IEX.
|
||||
|
||||
Symbols
|
||||
-------
|
||||
VOO: VANGUARD IX FUN/S&P 500 ETF
|
||||
GLD: SPDR Gold Trust
|
||||
VNQ: VANGUARD IX FUN/RL EST IX FD ETF
|
||||
VNQI: VANGUARD INTL E/GLB EX-US RL EST IX
|
||||
TLT: iShares Barclays 20+ Yr Treas.Bond
|
||||
TIP: iShares TIPS Bond ETF
|
||||
BNDX: VANGUARD CHARLO/TOTAL INTL BD ETF
|
||||
EEM: iShares MSCI Emerging Markets Indx
|
||||
RJI: Rogers International Commodity Index
|
||||
"""
|
||||
|
||||
if not start or not end:
|
||||
start = date(2015, 1, 1)
|
||||
end = date(2015, 12, 31)
|
||||
|
||||
symbols = ["VOO", "GLD", "VNQ", "VNQI", "TLT", "TIP", "BNDX", "EEM", "RJI"]
|
||||
|
||||
# Write headers
|
||||
full_path = os.path.join(data_dir, filename)
|
||||
with open(full_path, "w+") as f:
|
||||
f.write("date,symbol,open,high,low,close,volume\n")
|
||||
|
||||
columns = ["symbol", "open", "high", "low", "close", "volume"]
|
||||
for symbol in symbols:
|
||||
data = web.DataReader(symbol, "iex", start, end)
|
||||
data["symbol"] = symbol
|
||||
data.to_csv(
|
||||
full_path,
|
||||
mode="a",
|
||||
index_label="date",
|
||||
columns=columns,
|
||||
header=False)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
data_dir = get_data_dir()
|
||||
create_test_data(data_dir)
|
||||
create_synthetic_data(data_dir)
|
||||
fetch_balanced_data(data_dir)
|
||||
@@ -0,0 +1,12 @@
|
||||
import os
|
||||
import backtester as bt
|
||||
from backtester.utils import get_data_dir
|
||||
|
||||
data_dir = get_data_dir()
|
||||
spx_test_data = os.path.join(data_dir, "SPX_2008-2018.csv")
|
||||
|
||||
portfolio = bt.run(spx_test_data)
|
||||
report = portfolio.create_report()
|
||||
|
||||
print("Running Benchmark strategy on SPX data for 2008-2018")
|
||||
print(report.tail(30))
|
||||
@@ -0,0 +1,55 @@
|
||||
import unittest
|
||||
import os
|
||||
import backtester as bt
|
||||
from backtester.utils import get_data_dir
|
||||
|
||||
|
||||
class TestBalancedPortfolio(unittest.TestCase):
|
||||
"""Tests benchmark strategy using synthetic data"""
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
data_dir = get_data_dir()
|
||||
balanced_file = os.path.join(data_dir, "balanced_2015.csv")
|
||||
cls.port = bt.run(balanced_file)
|
||||
cls.report = cls.port.create_report()
|
||||
cls.initial_capital = 1000000
|
||||
cls.symbols = [
|
||||
"VOO", "GLD", "VNQ", "VNQI", "TLT", "TIP", "BNDX", "EEM", "RJI"
|
||||
]
|
||||
|
||||
def test_first_day_allocation(self):
|
||||
"""First day allocation of cash should be equal to initial capital"""
|
||||
self.assertEqual(TestBalancedPortfolio.report["Cash"].iloc[0],
|
||||
TestBalancedPortfolio.initial_capital)
|
||||
|
||||
def test_total_return(self):
|
||||
self.assertAlmostEqual(
|
||||
TestBalancedPortfolio.report["% Price"][-1], -0.041, delta=0.001)
|
||||
|
||||
def test_voo_allocation(self):
|
||||
"""Default VOO allocation should equal 30%"""
|
||||
self.assertAlmostEqual(
|
||||
TestBalancedPortfolio.report["VOO Exposure"].iloc[1],
|
||||
TestBalancedPortfolio.initial_capital * 0.3,
|
||||
delta=TestBalancedPortfolio.initial_capital * 0.01)
|
||||
|
||||
def test_voo_amount_constant(self):
|
||||
"""Default VOO amount should remain constant throughout backtest"""
|
||||
voo_amounts = TestBalancedPortfolio.report["VOO Amount"]
|
||||
self.assertTrue(all(voo_amounts[1:] == voo_amounts.iloc[1]))
|
||||
|
||||
def test_sum_of_all_symbols(self):
|
||||
"""Sum of all allocations should equal initial capital"""
|
||||
total_allocations = sum([
|
||||
TestBalancedPortfolio.report[symbol + " Exposure"].iloc[2]
|
||||
for symbol in TestBalancedPortfolio.symbols
|
||||
])
|
||||
self.assertAlmostEqual(
|
||||
total_allocations,
|
||||
TestBalancedPortfolio.initial_capital,
|
||||
delta=TestBalancedPortfolio.initial_capital * 0.1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,32 @@
|
||||
import unittest
|
||||
import os
|
||||
import backtester as bt
|
||||
from backtester.utils import get_data_dir
|
||||
|
||||
|
||||
class TestPortfolio(unittest.TestCase):
|
||||
"""Tests benchmark strategy using synthetic data"""
|
||||
|
||||
def setUp(self):
|
||||
data_dir = get_data_dir()
|
||||
synth_file = os.path.join(data_dir, "synthetic_data.csv")
|
||||
self.port = bt.run(synth_file)
|
||||
self.report = self.port.create_report()
|
||||
|
||||
def test_last_pct_price(self):
|
||||
self.assertAlmostEqual(self.report["% Price"].iloc[-1], 1999)
|
||||
|
||||
def test_mean_10_day_return(self):
|
||||
self.assertAlmostEqual(
|
||||
self.report["Interval Change"][:10].mean(), 0.31432, delta=0.0001)
|
||||
|
||||
def test_mean_100_day_return(self):
|
||||
self.assertAlmostEqual(
|
||||
self.report["Interval Change"][:100].mean(), 0.05229, delta=0.0001)
|
||||
|
||||
def test_last_total(self):
|
||||
self.assertEqual(self.report["Total Portfolio"].iloc[-1], 2000000000)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,15 @@
|
||||
import os
|
||||
|
||||
|
||||
def get_data_dir():
|
||||
"""Reads data path from environment variable $OPTIONS_DATA_PATH.
|
||||
If it is not set, defaults to `data/`
|
||||
"""
|
||||
|
||||
if "OPTIONS_DATA_PATH" in os.environ:
|
||||
data_dir = os.path.expanduser(os.environ["OPTIONS_DATA_PATH"])
|
||||
else:
|
||||
data_dir = "data"
|
||||
os.mkdir(data_dir)
|
||||
|
||||
return data_dir
|
||||
Reference in New Issue
Block a user