felixm
/
ML4T
1
0
Fork 0
Code Releases Activity

Compare commits

base: felixm:0519ae9336253afb2e641687b118cee747ee1755
Branches Tags
felixm:master
...
compare: felixm:10d87aefd39a35e0608518d8f97332de6d5f8268
Branches Tags
felixm:master

3 Commits
0519ae9336 ... 10d87aefd3

Author SHA1 Message Date
Felix Martin 10d87aefd3 Add tree learners to strategy evaluation directory 2020-11-04 15:15:24 -05:00
Felix Martin 05db89e8c2 Implement first version of strategy learner 
This version does not pass the automatic test.
 2020-11-04 15:14:27 -05:00
Felix Martin c40ffcf84b Show both MACD and indicator strat on figure 
Prepare for strategy learner.
 2020-11-04 09:23:42 -05:00
9 changed files with 634 additions and 435 deletions
Show Stats Expand all files Collapse all files

77
strategy_evaluation/AbstractTreeLearner.py Normal file
View File

@ -0,0 +1,77 @@
import numpy as np
class AbstractTreeLearner:
LEAF = -1
NA = -1
def author(self):
return 'felixm' # replace tb34 with your Georgia Tech username
def create_node(self, factor, split_value, left, right):
return np.array([(factor, split_value, left, right), ],
dtype='|i4, f4, i4, i4')
def query_point(self, point):
node_index = 0
while self.rel_tree[node_index][0] != self.LEAF:
node = self.rel_tree[node_index]
split_factor = node[0]
split_value = node[1]
if point[split_factor] <= split_value:
# Recurse into left sub-tree.
node_index += node[2]
else:
node_index += node[3]
v = self.rel_tree[node_index][1]
return v
def query(self, points):
"""
@summary: Estimate a set of test points given the model we built.
@param points: should be a numpy array with each row corresponding to a specific query.
@returns the estimated values according to the saved model.
"""
query_point = lambda p: self.query_point(p)
r = np.apply_along_axis(query_point, 1, points)
return r
def build_tree(self, xs, y):
"""
@summary: Build a decision tree from the training data.
@param dataX: X values of data to add
@param dataY: the Y training values
"""
assert(xs.shape[0] == y.shape[0])
assert(xs.shape[0] > 0) # If this is 0 something went wrong.
if xs.shape[0] <= self.leaf_size:
value = np.mean(y)
if value < -0.3:
value = -1
elif value > 0.3:
value = 1
else:
value = 0
return self.create_node(self.LEAF, value, self.NA, self.NA)
if np.all(y[0] == y):
return self.create_node(self.LEAF, y[0], self.NA, self.NA)
i, split_value = self.get_i_and_split_value(xs, y)
select_l = xs[:, i] <= split_value
select_r = xs[:, i] > split_value
lt = self.build_tree(xs[select_l], y[select_l])
rt = self.build_tree(xs[select_r], y[select_r])
root = self.create_node(i, split_value, 1, lt.shape[0] + 1)
root = np.concatenate([root, lt, rt])
return root
def addEvidence(self, data_x, data_y):
"""
@summary: Add training data to learner
@param dataX: X values of data to add
@param dataY: the Y training values
"""
self.rel_tree = self.build_tree(data_x, data_y)

47
strategy_evaluation/BagLearner.py Normal file
View File

@ -0,0 +1,47 @@
import numpy as np
from AbstractTreeLearner import AbstractTreeLearner
class BagLearner(AbstractTreeLearner):
def __init__(self, learner, bags=9, boost=False, verbose=False, kwargs={}):
self.learner = learner
self.verbose = verbose
self.bags = bags
self.learners = [learner(**kwargs) for _ in range(bags)]
def get_bag(self, data_x, data_y):
num_items = int(data_x.shape[0] * 0.5) # 50% of samples
bag_x, bag_y = [], []
for _ in range(num_items):
i = np.random.randint(0, data_x.shape[0])
bag_x.append(data_x[i,:])
bag_y.append(data_y[i])
return np.array(bag_x), np.array(bag_y)
def addEvidence(self, data_x, data_y):
"""
@summary: Add training data to learner
@param dataX: X values of data to add
@param dataY: the Y training values
"""
for learner in self.learners:
x, y = self.get_bag(data_x, data_y)
learner.addEvidence(x, y)
def query(self, points):
"""
@summary: Estimate a set of test points given the model we built.
@param points: numpy array with each row corresponding to a query.
@returns the estimated values according to the saved model.
"""
def to_discret(m):
print(m)
if m < -0.5:
return -1
elif m > 0.5:
return 1
return 0
m = np.mean([l.query(points) for l in self.learners], axis=0)
return m
# return np.apply_along_axis(to_discret, 1, m)

22
strategy_evaluation/ManualStrategy.py
View File

@ -36,18 +36,21 @@ class ManualStrategy:
print(volume)
def macd_strat(self, macd, orders):
"""Strategy based on MACD cross."""
def strat(ser):
m = macd.loc[ser.index]
prev_macd, prev_signal = m.iloc[0]
cur_macd, cur_signal = m.iloc[1]
prev_macd, prev_signal, _ = m.iloc[0]
cur_macd, cur_signal, _ = m.iloc[1]
shares = 0
if cur_macd < -1 and prev_macd < prev_signal and cur_macd > cur_signal:
if cur_macd < -1 and prev_macd < prev_signal \
and cur_macd > cur_signal:
if self.holding == 0:
shares = 1000
elif self.holding == -1000:
shares = 2000
elif cur_macd > 1 and prev_macd > prev_signal and cur_macd < cur_signal:
elif cur_macd > 1 and prev_macd > prev_signal \
and cur_macd < cur_signal:
if self.holding == 0:
shares = -1000
elif self.holding == 1000:
@ -58,6 +61,8 @@ class ManualStrategy:
orders['Shares'] = orders['Shares'].rolling(2).apply(strat)
def three_indicator_strat(self, macd, rsi, price_sma, orders):
"""Strategy based on three indicators. Thresholds selected based on
scatter plots."""
def strat(row):
shares = 0
_, _, macd_diff = macd.loc[row.name]
@ -87,7 +92,7 @@ class ManualStrategy:
def testPolicy(self, symbol="IBM",
sd=dt.datetime(2009, 1, 1),
ed=dt.datetime(2010, 1, 1),
sv=10000):
sv=10000, macd_strat=False):
self.holding = 0
df = util.get_data([symbol], pd.date_range(sd, ed))
@ -102,7 +107,8 @@ class ManualStrategy:
rsi = indicators.rsi(df, symbol)
price_sma = indicators.price_sma(df, symbol, [8])
# self.macd_strat(macd, orders)
self.three_indicator_strat(macd, rsi, price_sma, orders)
if macd_strat:
self.macd_strat(macd, orders)
else:
self.three_indicator_strat(macd, rsi, price_sma, orders)
return orders

30
strategy_evaluation/RTLearner.py Normal file
View File

@ -0,0 +1,30 @@
import numpy as np
from AbstractTreeLearner import AbstractTreeLearner
class RTLearner(AbstractTreeLearner):
def __init__(self, leaf_size = 1, verbose = False):
self.leaf_size = leaf_size
self.verbose = verbose
def get_i_and_split_value(self, xs, y):
"""
@summary: Pick a random i and split value.
Make sure that not all X are the same for i and also pick
different values to average the split_value from.
"""
i = np.random.randint(0, xs.shape[1])
while np.all(xs[0,i] == xs[:,i]):
i = np.random.randint(0, xs.shape[1])
# I don't know about the performance of this, but at least it
# terminates reliably. If the two elements are the same something is
# wrong.
a = np.array(list(set(xs[:, i])))
r1, r2 = np.random.choice(a, size = 2, replace = False)
assert(r1 != r2)
split_value = (r1 + r2) / 2.0
return i, split_value

154
strategy_evaluation/StrategyLearner.py
View File

@ -1,88 +1,94 @@
"""
Template for implementing StrategyLearner (c) 2016 Tucker Balch
Copyright 2018, Georgia Institute of Technology (Georgia Tech)
Atlanta, Georgia 30332
All Rights Reserved
Template code for CS 4646/7646
Georgia Tech asserts copyright ownership of this template and all derivative
works, including solutions to the projects assigned in this course. Students
and other users of this template code are advised not to share it with others
or to make it available on publicly viewable websites including repositories
such as github and gitlab. This copyright statement should not be removed
or edited.
We do grant permission to share solutions privately with non-students such
as potential employers. However, sharing with other current or future
students of CS 7646 is prohibited and subject to being investigated as a
GT honor code violation.
-----do not edit anything above this line---
Student Name: Tucker Balch (replace with your name)
GT User ID: tb34 (replace with your User ID)
GT ID: 900897987 (replace with your GT ID)
"""
import datetime as dt
import pandas as pd
import util as ut
import util
import indicators
from BagLearner import BagLearner
from RTLearner import RTLearner
class StrategyLearner(object):
# constructor
def __init__(self, verbose = False, impact=0.0, commission=0.0):
def __init__(self, verbose=False, impact=0.0, commission=0.0, testing=False):
self.verbose = verbose
self.impact = impact
self.commission = commission
self.testing = testing
# this method should create a QLearner, and train it for trading
def addEvidence(self, symbol = "IBM", \
sd=dt.datetime(2008,1,1), \
ed=dt.datetime(2009,1,1), \
sv = 10000):
# add your code to do learning here
# example usage of the old backward compatible util function
syms=[symbol]
dates = pd.date_range(sd, ed)
prices_all = ut.get_data(syms, dates) # automatically adds SPY
prices = prices_all[syms] # only portfolio symbols
# prices_SPY = prices_all['SPY'] # only SPY, for comparison later
if self.verbose: print(prices)
# example use with new colname
volume_all = ut.get_data(syms, dates, colname = "Volume") # automatically adds SPY
def _get_volume(self):
"""For reference."""
volume_all = ut.get_data(syms, dates, colname="Volume")
volume = volume_all[syms] # only portfolio symbols
# volume_SPY = volume_all['SPY'] # only SPY, for comparison later
if self.verbose: print(volume)
if self.verbose:
print(volume)
# this method should use the existing policy and test it against new data
def testPolicy(self, symbol = "IBM", \
sd=dt.datetime(2009,1,1), \
ed=dt.datetime(2010,1,1), \
sv = 10000):
def _add_indicators(self, df, symbol):
"""Add indicators for learning to DataFrame."""
df.drop(columns=["SPY"], inplace=True)
indicators.macd(df, symbol)
indicators.rsi(df, symbol)
indicators.price_sma(df, symbol, [8])
indicators.price_delta(df, symbol, 3)
df.dropna(inplace=True)
# here we build a fake set of trades
# your code should return the same sort of data
dates = pd.date_range(sd, ed)
prices_all = ut.get_data([symbol], dates) # automatically adds SPY
trades = prices_all[[symbol,]] # only portfolio symbols
# trades_SPY = prices_all['SPY'] # only SPY, for comparison later
trades.values[:,:] = 0 # set them all to nothing
trades.values[0,:] = 1000 # add a BUY at the start
trades.values[40,:] = -1000 # add a SELL
trades.values[41,:] = 1000 # add a BUY
trades.values[60,:] = -2000 # go short from long
trades.values[61,:] = 2000 # go long from short
trades.values[-1,:] = -1000 #exit on the last day
if self.verbose: print(type(trades)) # it better be a DataFrame!
if self.verbose: print(trades)
if self.verbose: print(prices_all)
return trades
def addEvidence(self, symbol="IBM",
sd=dt.datetime(2008, 1, 1),
ed=dt.datetime(2009, 1, 1),
sv=10000):
self.indicators = ['macd_diff', 'rsi', 'price_sma_8']
df = util.get_data([symbol], pd.date_range(sd, ed))
self._add_indicators(df, symbol)
def classify_y(row):
if row > 0.1:
return 1
elif row < -0.1:
return -1
return 0
self.learner = RTLearner(leaf_size = 7)
# self.learner = BagLearner(RTLearner, 5, {'leaf_size': 5})
data_x = df[self.indicators].to_numpy()
y = df['pct_3'].apply(classify_y)
self.learner.addEvidence(data_x, y.to_numpy())
return y
def strat(self, data_y, orders):
self.holding = 0
def strat(row):
y = int(data_y.loc[row.name][0])
shares = 0
if self.holding == 0 and y == 1:
shares = 1000
elif self.holding == -1000 and y == 1:
shares = 2000
elif self.holding == 0 and y == -1:
shares = -1000
elif self.holding == 1000 and y == -1:
shares = -2000
self.holding += shares
return shares
orders["Shares"] = orders.apply(strat, axis=1)
def testPolicy(self, symbol="IBM",
sd=dt.datetime(2009, 1, 1),
ed=dt.datetime(2010, 1, 1),
sv=10000):
df = util.get_data([symbol], pd.date_range(sd, ed))
self._add_indicators(df, symbol)
data_x = df[self.indicators].to_numpy()
data_y = pd.DataFrame(index=df.index, data=self.learner.query(data_x))
orders = pd.DataFrame(index=df.index)
orders["Symbol"] = symbol
orders["Order"] = ""
orders["Shares"] = 0
self.strat(data_y, orders)
if self.testing:
return orders
else:
return orders[["Shares"]]
if __name__=="__main__":
print("One does not simply think up a strategy")

82
strategy_evaluation/experiment1.py
View File

@ -9,6 +9,7 @@ import matplotlib.pyplot as plt
from matplotlib.widgets import MultiCursor
from BenchmarkStrategy import BenchmarkStrategy
from ManualStrategy import ManualStrategy
from StrategyLearner import StrategyLearner
def plot_indicators(symbol, df):
@ -16,7 +17,6 @@ def plot_indicators(symbol, df):
price_sma = indicators.price_sma(df, symbol, [8])
bb = indicators.bollinger_band(df, symbol)
sma = indicators.sma(df, symbol, [8])
rsi = indicators.rsi(df, symbol)
macd = indicators.macd(df, symbol).copy()
@ -57,41 +57,81 @@ def visualize_correlations(symbol, df):
sys.exit(0)
def experiment1():
symbol = "JPM"
start_value = 10000
sd = dt.datetime(2008, 1, 1) # in-sample
ed = dt.datetime(2009, 12, 31) # in-sample
# sd = dt.datetime(2010, 1, 1) # out-sample
# ed = dt.datetime(2011, 12, 31) # out-sample
def compare_manual_strategies(symbol, sv, sd, ed):
df = util.get_data([symbol], pd.date_range(sd, ed))
df.drop(columns=["SPY"], inplace=True)
# visualize_correlations(symbol, df)
# plot_indicators(symbol, df)
bs = BenchmarkStrategy()
orders = bs.testPolicy(symbol, sd, ed, start_value)
df["Benchmark"] = marketsim.compute_portvals(orders, start_value)
orders = bs.testPolicy(symbol, sd, ed, sv)
df["Benchmark"] = marketsim.compute_portvals(orders, sv)
df["Orders Benchmark"] = orders["Shares"]
ms = ManualStrategy()
orders = ms.testPolicy(symbol, sd, ed, start_value)
df["Manual"] = marketsim.compute_portvals(orders, start_value)
df["Orders Manual"] = orders["Shares"]
df["Holding Manual"] = orders["Shares"].cumsum()
orders = ms.testPolicy(symbol, sd, ed, sv, macd_strat=True)
df["MACD Strat"] = marketsim.compute_portvals(orders, sv)
df["Orders MACD"] = orders["Shares"]
# df["Holding Manual"] = orders["Shares"].cumsum()
orders = ms.testPolicy(symbol, sd, ed, sv)
df["Three Strat"] = marketsim.compute_portvals(orders, sv)
df["Orders Three"] = orders["Shares"]
fig, ax = plt.subplots(3, sharex=True)
df[[symbol]].plot(ax=ax[0])
df[["Benchmark", "Manual"]].plot(ax=ax[1])
df[["Orders Benchmark", "Orders Manual"]].plot(ax=ax[2])
df[["Benchmark", "MACD Strat", "Three Strat"]].plot(ax=ax[1])
df[["Orders Benchmark", "Orders MACD", "Orders Three"]].plot(ax=ax[2])
for a in ax:
a.grid()
multi = MultiCursor(fig.canvas, ax, color='r', lw=0.5)
MultiCursor(fig.canvas, ax, color='r', lw=0.5)
# plt.show()
fig.set_size_inches(10, 8, forward=True)
plt.savefig('figure_1.png', dpi=fig.dpi)
def experiment1():
symbol = "JPM"
sv = 10000
sd = dt.datetime(2008, 1, 1) # in-sample
ed = dt.datetime(2009, 12, 31) # in-sample
sd_out = dt.datetime(2010, 1, 1) # out-sample
ed_out = dt.datetime(2011, 12, 31) # out-sample
df = util.get_data([symbol], pd.date_range(sd, ed_out))
df.drop(columns=["SPY"], inplace=True)
# visualize_correlations(symbol, df)
# plot_indicators(symbol, df)
# compare_manual_strategies(symbol, sv, sd, ed)
bs = BenchmarkStrategy()
orders = bs.testPolicy(symbol, sd_out, ed_out, sv)
df["Benchmark"] = marketsim.compute_portvals(orders, sv)
df["Orders Benchmark"] = orders["Shares"]
sl = StrategyLearner(testing=True)
sl.addEvidence(symbol, sd, ed, sv)
orders = sl.testPolicy(symbol, sd_out, ed_out, sv)
df["SL"] = marketsim.compute_portvals(orders, sv)
df["Orders SL"] = orders["Shares"]
fig, ax = plt.subplots(3, sharex=True)
df[[symbol]].plot(ax=ax[0])
df[["Benchmark", "SL"]].plot(ax=ax[1])
df[["Orders Benchmark", "Orders SL"]].plot(ax=ax[2])
for a in ax:
a.grid()
MultiCursor(fig.canvas, ax, color='r', lw=0.5)
plt.show()
# plt.savefig('figure_1.png')
# For debugging the classification learner:
# df["y_train"] = sl.addEvidence(symbol, sd, ed, sv)
# df["y_query"] = sl.testPolicy(symbol, sd, ed, sv)
# df[["y_train", "y_query"]].plot(ax=ax[1])
if __name__ == "__main__":

BIN
strategy_evaluation/figure_1.png
View File

Binary file not shown.
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 68 KiB

After

Width:  |  Height:  |  Size: 112 KiB

648
strategy_evaluation/grade_strategy_learner.py
View File

@ -1,242 +1,242 @@
"""MC3-P3: Strategy Learner - grading script.
Usage:
- Switch to a student feedback directory first (will write "points.txt" and "comments.txt" in pwd).
- Run this script with both ml4t/ and student solution in PYTHONPATH, e.g.:
PYTHONPATH=ml4t:MC1-P2/jdoe7 python ml4t/mc2_p1_grading/grade_marketsim.py
Copyright 2017, Georgia Tech Research Corporation
Atlanta, Georgia 30332-0415
All Rights Reserved
Template code for CS 4646/7646
Georgia Tech asserts copyright ownership of this template and all derivative
works, including solutions to the projects assigned in this course. Students
and other users of this template code are advised not to share it with others
or to make it available on publicly viewable websites including repositories
such as github and gitlab. This copyright statement should not be removed
or edited.
We do grant permission to share solutions privately with non-students such
as potential employers. However, sharing with other current or future
students of CS 7646 is prohibited and subject to being investigated as a
GT honor code violation.
-----do not edit anything above this line---
Student Name: Tucker Balch (replace with your name)
GT User ID: tb34 (replace with your User ID)
GT ID: 900897987 (replace with your GT ID)
"""
import pytest
from grading.grading import grader, GradeResult, run_with_timeout, IncorrectOutput
import os
import sys
import traceback as tb
import datetime as dt
import numpy as np
import pandas as pd
from collections import namedtuple
import time
import util
import random
# Test cases
StrategyTestCase = namedtuple('Strategy', ['description','insample_args','outsample_args','benchmark_type','benchmark','impact','train_time','test_time','max_time','seed'])
strategy_test_cases = [
StrategyTestCase(
description="ML4T-220",
insample_args=dict(symbol="ML4T-220",sd=dt.datetime(2008,1,1),ed=dt.datetime(2009,12,31),sv=100000),
outsample_args=dict(symbol="ML4T-220",sd=dt.datetime(2010,1,1),ed=dt.datetime(2011,12,31),sv=100000),
benchmark_type='clean',
benchmark=1.0, #benchmark updated Apr 24 2017
impact=0.0,
train_time=25,
test_time=5,
max_time=60,
seed=1481090000
),
StrategyTestCase(
description="AAPL",
insample_args=dict(symbol="AAPL",sd=dt.datetime(2008,1,1),ed=dt.datetime(2009,12,31),sv=100000),
outsample_args=dict(symbol="AAPL",sd=dt.datetime(2010,1,1),ed=dt.datetime(2011,12,31),sv=100000),
benchmark_type='stock',
benchmark=0.1581999999999999, #benchmark computed Nov 22 2017
impact=0.0,
train_time=25,
test_time=5,
max_time=60,
seed=1481090000
),
StrategyTestCase(
description="SINE_FAST_NOISE",
insample_args=dict(symbol="SINE_FAST_NOISE",sd=dt.datetime(2008,1,1),ed=dt.datetime(2009,12,31),sv=100000),
outsample_args=dict(symbol="SINE_FAST_NOISE",sd=dt.datetime(2010,1,1),ed=dt.datetime(2011,12,31),sv=100000),
benchmark_type='noisy',
benchmark=2.0, #benchmark updated Apr 24 2017
impact=0.0,
train_time=25,
test_time=5,
max_time=60,
seed=1481090000
),
StrategyTestCase(
description="UNH - In sample",
insample_args=dict(symbol="UNH",sd=dt.datetime(2008,1,1),ed=dt.datetime(2009,12,31),sv=100000),
outsample_args=dict(symbol="UNH",sd=dt.datetime(2010,1,1),ed=dt.datetime(2011,12,31),sv=100000),
benchmark_type='stock',
benchmark= -0.25239999999999996, #benchmark computed Nov 22 2017
impact=0.0,
train_time=25,
test_time=5,
max_time=60,
seed=1481090000
),
]
max_points = 60.0
html_pre_block = True # surround comments with HTML <pre> tag (for T-Square comments field)
MAX_HOLDINGS = 1000
# Test functon(s)
@pytest.mark.parametrize("description, insample_args, outsample_args, benchmark_type, benchmark, impact, train_time, test_time, max_time, seed", strategy_test_cases)
def test_strategy(description, insample_args, outsample_args, benchmark_type, benchmark, impact, train_time, test_time, max_time, seed, grader):
"""Test StrategyLearner.
Requires test description, insample args (dict), outsample args (dict), benchmark_type (str), benchmark (float)
max time (seconds), points for this test case (int), random seed (long), and a grader fixture.
"""
points_earned = 0.0 # initialize points for this test case
try:
incorrect = True
if not 'StrategyLearner' in globals():
import importlib
m = importlib.import_module('StrategyLearner')
globals()['StrategyLearner'] = m
outsample_cr_to_beat = None
if benchmark_type == 'clean':
outsample_cr_to_beat = benchmark
def timeoutwrapper_strategylearner():
#Set fixed seed for repetability
np.random.seed(seed)
random.seed(seed)
learner = StrategyLearner.StrategyLearner(verbose=False,impact=impact)
tmp = time.time()
learner.addEvidence(**insample_args)
train_t = time.time()-tmp
tmp = time.time()
insample_trades_1 = learner.testPolicy(**insample_args)
test_t = time.time()-tmp
insample_trades_2 = learner.testPolicy(**insample_args)
tmp = time.time()
outsample_trades = learner.testPolicy(**outsample_args)
out_test_t = time.time()-tmp
return insample_trades_1, insample_trades_2, outsample_trades, train_t, test_t, out_test_t
msgs = []
in_trades_1, in_trades_2, out_trades, train_t, test_t, out_test_t = run_with_timeout(timeoutwrapper_strategylearner,max_time,(),{})
incorrect = False
if len(in_trades_1.shape)!=2 or in_trades_1.shape[1]!=1:
incorrect=True
msgs.append(" First insample trades DF has invalid shape: {}".format(in_trades_1.shape))
elif len(in_trades_2.shape)!=2 or in_trades_2.shape[1]!=1:
incorrect=True
msgs.append(" Second insample trades DF has invalid shape: {}".format(in_trades_2.shape))
elif len(out_trades.shape)!=2 or out_trades.shape[1]!=1:
incorrect=True
msgs.append(" Out-of-sample trades DF has invalid shape: {}".format(out_trades.shape))
else:
tmp_csum=0.0
for date,trade in in_trades_1.iterrows():
tmp_csum+= trade.iloc[0]
"""MC3-P3: Strategy Learner - grading script.
Usage:
- Switch to a student feedback directory first (will write "points.txt" and "comments.txt" in pwd).
- Run this script with both ml4t/ and student solution in PYTHONPATH, e.g.:
PYTHONPATH=ml4t:MC1-P2/jdoe7 python ml4t/mc2_p1_grading/grade_marketsim.py
Copyright 2017, Georgia Tech Research Corporation
Atlanta, Georgia 30332-0415
All Rights Reserved
Template code for CS 4646/7646
Georgia Tech asserts copyright ownership of this template and all derivative
works, including solutions to the projects assigned in this course. Students
and other users of this template code are advised not to share it with others
or to make it available on publicly viewable websites including repositories
such as github and gitlab. This copyright statement should not be removed
or edited.
We do grant permission to share solutions privately with non-students such
as potential employers. However, sharing with other current or future
students of CS 7646 is prohibited and subject to being investigated as a
GT honor code violation.
-----do not edit anything above this line---
Student Name: Tucker Balch (replace with your name)
GT User ID: tb34 (replace with your User ID)
GT ID: 900897987 (replace with your GT ID)
"""
import pytest
from grading.grading import grader, GradeResult, run_with_timeout, IncorrectOutput
import os
import sys
import traceback as tb
import datetime as dt
import numpy as np
import pandas as pd
from collections import namedtuple
import time
import util
import random
# Test cases
StrategyTestCase = namedtuple('Strategy', ['description','insample_args','outsample_args','benchmark_type','benchmark','impact','train_time','test_time','max_time','seed'])
strategy_test_cases = [
StrategyTestCase(
description="ML4T-220",
insample_args=dict(symbol="ML4T-220",sd=dt.datetime(2008,1,1),ed=dt.datetime(2009,12,31),sv=100000),
outsample_args=dict(symbol="ML4T-220",sd=dt.datetime(2010,1,1),ed=dt.datetime(2011,12,31),sv=100000),
benchmark_type='clean',
benchmark=1.0, #benchmark updated Apr 24 2017
impact=0.0,
train_time=25,
test_time=5,
max_time=60,
seed=1481090000
),
StrategyTestCase(
description="AAPL",
insample_args=dict(symbol="AAPL",sd=dt.datetime(2008,1,1),ed=dt.datetime(2009,12,31),sv=100000),
outsample_args=dict(symbol="AAPL",sd=dt.datetime(2010,1,1),ed=dt.datetime(2011,12,31),sv=100000),
benchmark_type='stock',
benchmark=0.1581999999999999, #benchmark computed Nov 22 2017
impact=0.0,
train_time=25,
test_time=5,
max_time=60,
seed=1481090000
),
StrategyTestCase(
description="SINE_FAST_NOISE",
insample_args=dict(symbol="SINE_FAST_NOISE",sd=dt.datetime(2008,1,1),ed=dt.datetime(2009,12,31),sv=100000),
outsample_args=dict(symbol="SINE_FAST_NOISE",sd=dt.datetime(2010,1,1),ed=dt.datetime(2011,12,31),sv=100000),
benchmark_type='noisy',
benchmark=2.0, #benchmark updated Apr 24 2017
impact=0.0,
train_time=25,
test_time=5,
max_time=60,
seed=1481090000
),
StrategyTestCase(
description="UNH - In sample",
insample_args=dict(symbol="UNH",sd=dt.datetime(2008,1,1),ed=dt.datetime(2009,12,31),sv=100000),
outsample_args=dict(symbol="UNH",sd=dt.datetime(2010,1,1),ed=dt.datetime(2011,12,31),sv=100000),
benchmark_type='stock',
benchmark= -0.25239999999999996, #benchmark computed Nov 22 2017
impact=0.0,
train_time=25,
test_time=5,
max_time=60,
seed=1481090000
),
]
max_points = 60.0
html_pre_block = True # surround comments with HTML <pre> tag (for T-Square comments field)
MAX_HOLDINGS = 1000
# Test functon(s)
@pytest.mark.parametrize("description, insample_args, outsample_args, benchmark_type, benchmark, impact, train_time, test_time, max_time, seed", strategy_test_cases)
def test_strategy(description, insample_args, outsample_args, benchmark_type, benchmark, impact, train_time, test_time, max_time, seed, grader):
"""Test StrategyLearner.
Requires test description, insample args (dict), outsample args (dict), benchmark_type (str), benchmark (float)
max time (seconds), points for this test case (int), random seed (long), and a grader fixture.
"""
points_earned = 0.0 # initialize points for this test case
try:
incorrect = True
if not 'StrategyLearner' in globals():
import importlib
m = importlib.import_module('StrategyLearner')
globals()['StrategyLearner'] = m
outsample_cr_to_beat = None
if benchmark_type == 'clean':
outsample_cr_to_beat = benchmark
def timeoutwrapper_strategylearner():
#Set fixed seed for repetability
np.random.seed(seed)
random.seed(seed)
learner = StrategyLearner.StrategyLearner(verbose=False,impact=impact)
tmp = time.time()
learner.addEvidence(**insample_args)
train_t = time.time()-tmp
tmp = time.time()
insample_trades_1 = learner.testPolicy(**insample_args)
test_t = time.time()-tmp
insample_trades_2 = learner.testPolicy(**insample_args)
tmp = time.time()
outsample_trades = learner.testPolicy(**outsample_args)
out_test_t = time.time()-tmp
return insample_trades_1, insample_trades_2, outsample_trades, train_t, test_t, out_test_t
msgs = []
in_trades_1, in_trades_2, out_trades, train_t, test_t, out_test_t = run_with_timeout(timeoutwrapper_strategylearner,max_time,(),{})
incorrect = False
if len(in_trades_1.shape)!=2 or in_trades_1.shape[1]!=1:
incorrect=True
msgs.append(" First insample trades DF has invalid shape: {}".format(in_trades_1.shape))
elif len(in_trades_2.shape)!=2 or in_trades_2.shape[1]!=1:
incorrect=True
msgs.append(" Second insample trades DF has invalid shape: {}".format(in_trades_2.shape))
elif len(out_trades.shape)!=2 or out_trades.shape[1]!=1:
incorrect=True
msgs.append(" Out-of-sample trades DF has invalid shape: {}".format(out_trades.shape))
else:
tmp_csum=0.0
for date,trade in in_trades_1.iterrows():
tmp_csum+= trade.iloc[0]
if (trade.iloc[0]!=0) and\
(trade.abs().iloc[0]!=MAX_HOLDINGS) and\
(trade.abs().iloc[0]!=2*MAX_HOLDINGS):
incorrect=True
msgs.append(" illegal trade in first insample DF. abs(trade) not one of ({},{},{}).\n Date {}, Trade {}".format(0,MAX_HOLDINGS,2*MAX_HOLDINGS,date,trade))
break
elif abs(tmp_csum)>MAX_HOLDINGS:
incorrect=True
msgs.append(" holdings more than {} long or short in first insample DF. Date {}, Trade {}".format(MAX_HOLDINGS,date,trade))
break
tmp_csum=0.0
for date,trade in in_trades_2.iterrows():
tmp_csum+= trade.iloc[0]
(trade.abs().iloc[0]!=2*MAX_HOLDINGS):
incorrect=True
msgs.append(" illegal trade in first insample DF. abs(trade) not one of ({},{},{}).\n Date {}, Trade {}".format(0,MAX_HOLDINGS,2*MAX_HOLDINGS,date,trade))
break
elif abs(tmp_csum)>MAX_HOLDINGS:
incorrect=True
msgs.append(" holdings more than {} long or short in first insample DF. Date {}, Trade {}".format(MAX_HOLDINGS,date,trade))
break
tmp_csum=0.0
for date,trade in in_trades_2.iterrows():
tmp_csum+= trade.iloc[0]
if (trade.iloc[0]!=0) and\
(trade.abs().iloc[0]!=MAX_HOLDINGS) and\
(trade.abs().iloc[0]!=2*MAX_HOLDINGS):
incorrect=True
msgs.append(" illegal trade in second insample DF. abs(trade) not one of ({},{},{}).\n Date {}, Trade {}".format(0,MAX_HOLDINGS,2*MAX_HOLDINGS,date,trade))
break
elif abs(tmp_csum)>MAX_HOLDINGS:
incorrect=True
msgs.append(" holdings more than {} long or short in second insample DF. Date {}, Trade {}".format(MAX_HOLDINGS,date,trade))
break
tmp_csum=0.0
for date,trade in out_trades.iterrows():
tmp_csum+= trade.iloc[0]
(trade.abs().iloc[0]!=2*MAX_HOLDINGS):
incorrect=True
msgs.append(" illegal trade in second insample DF. abs(trade) not one of ({},{},{}).\n Date {}, Trade {}".format(0,MAX_HOLDINGS,2*MAX_HOLDINGS,date,trade))
break
elif abs(tmp_csum)>MAX_HOLDINGS:
incorrect=True
msgs.append(" holdings more than {} long or short in second insample DF. Date {}, Trade {}".format(MAX_HOLDINGS,date,trade))
break
tmp_csum=0.0
for date,trade in out_trades.iterrows():
tmp_csum+= trade.iloc[0]
if (trade.iloc[0]!=0) and\
(trade.abs().iloc[0]!=MAX_HOLDINGS) and\
(trade.abs().iloc[0]!=2*MAX_HOLDINGS):
incorrect=True
msgs.append(" illegal trade in out-of-sample DF. abs(trade) not one of ({},{},{}).\n Date {}, Trade {}".format(0,MAX_HOLDINGS,2*MAX_HOLDINGS,date,trade))
break
elif abs(tmp_csum)>MAX_HOLDINGS:
incorrect=True
msgs.append(" holdings more than {} long or short in out-of-sample DF. Date {}, Trade {}".format(MAX_HOLDINGS,date,trade))
break
(trade.abs().iloc[0]!=2*MAX_HOLDINGS):
incorrect=True
msgs.append(" illegal trade in out-of-sample DF. abs(trade) not one of ({},{},{}).\n Date {}, Trade {}".format(0,MAX_HOLDINGS,2*MAX_HOLDINGS,date,trade))
break
elif abs(tmp_csum)>MAX_HOLDINGS:
incorrect=True
msgs.append(" holdings more than {} long or short in out-of-sample DF. Date {}, Trade {}".format(MAX_HOLDINGS,date,trade))
break
# if (((in_trades_1.abs()!=0) & (in_trades_1.abs()!=MAX_HOLDINGS) & (in_trades_1.abs()!=2*MAX_HOLDINGS)).any().any() or\
# ((in_trades_2.abs()!=0) & (in_trades_2.abs()!=MAX_HOLDINGS) & (in_trades_2.abs()!=2*MAX_HOLDINGS)).any().any() or\
# ((out_trades.abs()!=0) & (out_trades.abs()!=MAX_HOLDINGS) & (out_trades.abs()!=2*MAX_HOLDINGS)).any().any()):
# incorrect = True
# msgs.append(" illegal trade. abs(trades) not one of ({},{},{})".format(0,MAX_HOLDINGS,2*MAX_HOLDINGS))
# if ((in_trades_1.cumsum().abs()>MAX_HOLDINGS).any()[0]) or ((in_trades_2.cumsum().abs()>MAX_HOLDINGS).any()[0]) or ((out_trades.cumsum().abs()>MAX_HOLDINGS).any()[0]):
# incorrect = True
# msgs.append(" holdings more than {} long or short".format(MAX_HOLDINGS))
if not(incorrect):
if train_t>train_time:
incorrect=True
msgs.append(" addEvidence() took {} seconds, max allowed {}".format(train_t,train_time))
else:
points_earned += 1.0
if test_t > test_time:
incorrect = True
msgs.append(" testPolicy() took {} seconds, max allowed {}".format(test_t,test_time))
else:
points_earned += 2.0
if not((in_trades_1 == in_trades_2).all()[0]):
incorrect = True
mismatches = in_trades_1.join(in_trades_2,how='outer',lsuffix='1',rsuffix='2')
mismatches = mismatches[mismatches.iloc[:,0]!=mismatches.iloc[:,1]]
msgs.append(" consecutive calls to testPolicy() with same input did not produce same output:")
msgs.append(" Mismatched trades:\n {}".format(mismatches))
else:
points_earned += 2.0
student_insample_cr = evalPolicy2(insample_args['symbol'],in_trades_1,insample_args['sv'],insample_args['sd'],insample_args['ed'],market_impact=impact,commission_cost=0.0)
student_outsample_cr = evalPolicy2(outsample_args['symbol'],out_trades, outsample_args['sv'],outsample_args['sd'],outsample_args['ed'],market_impact=impact,commission_cost=0.0)
if student_insample_cr <= benchmark:
incorrect = True
msgs.append(" in-sample return ({}) did not beat benchmark ({})".format(student_insample_cr,benchmark))
else:
points_earned += 5.0
if outsample_cr_to_beat is None:
if out_test_t > test_time:
incorrect = True
msgs.append(" out-sample took {} seconds, max of {}".format(out_test_t,test_time))
else:
points_earned += 5.0
else:
if student_outsample_cr < outsample_cr_to_beat:
incorrect = True
msgs.append(" out-sample return ({}) did not beat benchmark ({})".format(student_outsample_cr,outsample_cr_to_beat))
else:
points_earned += 5.0
if incorrect:
# ((out_trades.abs()!=0) & (out_trades.abs()!=MAX_HOLDINGS) & (out_trades.abs()!=2*MAX_HOLDINGS)).any().any()):
# incorrect = True
# msgs.append(" illegal trade. abs(trades) not one of ({},{},{})".format(0,MAX_HOLDINGS,2*MAX_HOLDINGS))
# if ((in_trades_1.cumsum().abs()>MAX_HOLDINGS).any()[0]) or ((in_trades_2.cumsum().abs()>MAX_HOLDINGS).any()[0]) or ((out_trades.cumsum().abs()>MAX_HOLDINGS).any()[0]):
# incorrect = True
# msgs.append(" holdings more than {} long or short".format(MAX_HOLDINGS))
if not(incorrect):
if train_t>train_time:
incorrect=True
msgs.append(" addEvidence() took {} seconds, max allowed {}".format(train_t,train_time))
else:
points_earned += 1.0
if test_t > test_time:
incorrect = True
msgs.append(" testPolicy() took {} seconds, max allowed {}".format(test_t,test_time))
else:
points_earned += 2.0
if not((in_trades_1 == in_trades_2).all()[0]):
incorrect = True
mismatches = in_trades_1.join(in_trades_2,how='outer',lsuffix='1',rsuffix='2')
mismatches = mismatches[mismatches.iloc[:,0]!=mismatches.iloc[:,1]]
msgs.append(" consecutive calls to testPolicy() with same input did not produce same output:")
msgs.append(" Mismatched trades:\n {}".format(mismatches))
else:
points_earned += 2.0
student_insample_cr = evalPolicy2(insample_args['symbol'],in_trades_1,insample_args['sv'],insample_args['sd'],insample_args['ed'],market_impact=impact,commission_cost=0.0)
student_outsample_cr = evalPolicy2(outsample_args['symbol'],out_trades, outsample_args['sv'],outsample_args['sd'],outsample_args['ed'],market_impact=impact,commission_cost=0.0)
if student_insample_cr <= benchmark:
incorrect = True
msgs.append(" in-sample return ({}) did not beat benchmark ({})".format(student_insample_cr,benchmark))
else:
points_earned += 5.0
if outsample_cr_to_beat is None:
if out_test_t > test_time:
incorrect = True
msgs.append(" out-sample took {} seconds, max of {}".format(out_test_t,test_time))
else:
points_earned += 5.0
else:
if student_outsample_cr < outsample_cr_to_beat:
incorrect = True
msgs.append(" out-sample return ({}) did not beat benchmark ({})".format(student_outsample_cr,outsample_cr_to_beat))
else:
points_earned += 5.0
if incorrect:
inputs_str = " insample_args: {}\n" \
" outsample_args: {}\n" \
" benchmark_type: {}\n" \
@ -244,96 +244,96 @@ def test_strategy(description, insample_args, outsample_args, benchmark_type, be
" train_time: {}\n" \
" test_time: {}\n" \
" max_time: {}\n" \
" seed: {}\n".format(insample_args, outsample_args, benchmark_type, benchmark, train_time, test_time, max_time,seed)
raise IncorrectOutput("Test failed on one or more output criteria.\n Inputs:\n{}\n Failures:\n{}".format(inputs_str, "\n".join(msgs)))
except Exception as e:
# Test result: failed
msg = "Test case description: {}\n".format(description)
# Generate a filtered stacktrace, only showing erroneous lines in student file(s)
tb_list = tb.extract_tb(sys.exc_info()[2])
for i in range(len(tb_list)):
row = tb_list[i]
tb_list[i] = (os.path.basename(row[0]), row[1], row[2], row[3]) # show only filename instead of long absolute path
# tb_list = [row for row in tb_list if row[0] in ['QLearner.py','StrategyLearner.py']]
if tb_list:
msg += "Traceback:\n"
msg += ''.join(tb.format_list(tb_list)) # contains newlines
elif 'grading_traceback' in dir(e):
msg += "Traceback:\n"
msg += ''.join(tb.format_list(e.grading_traceback))
msg += "{}: {}".format(e.__class__.__name__, str(e))
# Report failure result to grader, with stacktrace
grader.add_result(GradeResult(outcome='failed', points=points_earned, msg=msg))
raise
else:
# Test result: passed (no exceptions)
grader.add_result(GradeResult(outcome='passed', points=points_earned, msg=None))
def compute_benchmark(sd,ed,sv,symbol,market_impact,commission_cost,max_holdings):
date_idx = util.get_data([symbol,],pd.date_range(sd,ed)).index
orders = pd.DataFrame(index=date_idx)
orders['orders'] = 0; orders['orders'][0] = max_holdings; orders['orders'][-1] = -max_holdings
return evalPolicy2(symbol,orders,sv,sd,ed,market_impact,commission_cost)
def evalPolicy(student_trades,sym_prices,startval):
ending_cash = startval - student_trades.mul(sym_prices,axis=0).sum()
ending_stocks = student_trades.sum()*sym_prices.iloc[-1]
return float((ending_cash+ending_stocks)/startval)-1.0
def evalPolicy2(symbol, student_trades, startval, sd, ed, market_impact,commission_cost):
orders_df = pd.DataFrame(columns=['Shares','Order','Symbol'])
for row_idx in student_trades.index:
nshares = student_trades.loc[row_idx][0]
if nshares == 0:
continue
order = 'sell' if nshares < 0 else 'buy'
new_row = pd.DataFrame([[abs(nshares),order,symbol],],columns=['Shares','Order','Symbol'],index=[row_idx,])
orders_df = orders_df.append(new_row)
portvals = compute_portvals(orders_df, sd, ed, startval,market_impact,commission_cost)
return float(portvals[-1]/portvals[0])-1
def compute_portvals(orders_df, start_date, end_date, startval, market_impact=0.0, commission_cost=0.0):
"""Simulate the market for the given date range and orders file."""
symbols = []
orders = []
orders_df = orders_df.sort_index()
for date, order in orders_df.iterrows():
shares = order['Shares']
action = order['Order']
symbol = order['Symbol']
if action.lower() == 'sell':
shares *= -1
order = (date, symbol, shares)
orders.append(order)
symbols.append(symbol)
symbols = list(set(symbols))
dates = pd.date_range(start_date, end_date)
prices_all = util.get_data(symbols, dates)
prices = prices_all[symbols]
prices = prices.fillna(method='ffill').fillna(method='bfill')
prices['_CASH'] = 1.0
trades = pd.DataFrame(index=prices.index, columns=symbols)
trades = trades.fillna(0)
cash = pd.Series(index=prices.index)
cash = cash.fillna(0)
cash.iloc[0] = startval
for date, symbol, shares in orders:
price = prices[symbol][date]
val = shares * price
# transaction cost model
val += commission_cost + (pd.np.abs(shares)*price*market_impact)
positions = prices.loc[date] * trades.sum()
totalcash = cash.sum()
if (date < prices.index.min()) or (date > prices.index.max()):
continue
trades[symbol][date] += shares
cash[date] -= val
trades['_CASH'] = cash
holdings = trades.cumsum()
df_portvals = (prices * holdings).sum(axis=1)
return df_portvals
if __name__ == "__main__":
pytest.main(["-s", __file__])
" seed: {}\n".format(insample_args, outsample_args, benchmark_type, benchmark, train_time, test_time, max_time,seed)
raise IncorrectOutput("Test failed on one or more output criteria.\n Inputs:\n{}\n Failures:\n{}".format(inputs_str, "\n".join(msgs)))
except Exception as e:
# Test result: failed
msg = "Test case description: {}\n".format(description)
# Generate a filtered stacktrace, only showing erroneous lines in student file(s)
tb_list = tb.extract_tb(sys.exc_info()[2])
for i in range(len(tb_list)):
row = tb_list[i]
tb_list[i] = (os.path.basename(row[0]), row[1], row[2], row[3]) # show only filename instead of long absolute path
# tb_list = [row for row in tb_list if row[0] in ['QLearner.py','StrategyLearner.py']]
if tb_list:
msg += "Traceback:\n"
msg += ''.join(tb.format_list(tb_list)) # contains newlines
elif 'grading_traceback' in dir(e):
msg += "Traceback:\n"
msg += ''.join(tb.format_list(e.grading_traceback))
msg += "{}: {}".format(e.__class__.__name__, str(e))
# Report failure result to grader, with stacktrace
grader.add_result(GradeResult(outcome='failed', points=points_earned, msg=msg))
raise
else:
# Test result: passed (no exceptions)
grader.add_result(GradeResult(outcome='passed', points=points_earned, msg=None))
def compute_benchmark(sd,ed,sv,symbol,market_impact,commission_cost,max_holdings):
date_idx = util.get_data([symbol,],pd.date_range(sd,ed)).index
orders = pd.DataFrame(index=date_idx)
orders['orders'] = 0; orders['orders'][0] = max_holdings; orders['orders'][-1] = -max_holdings
return evalPolicy2(symbol,orders,sv,sd,ed,market_impact,commission_cost)
def evalPolicy(student_trades,sym_prices,startval):
ending_cash = startval - student_trades.mul(sym_prices,axis=0).sum()
ending_stocks = student_trades.sum()*sym_prices.iloc[-1]
return float((ending_cash+ending_stocks)/startval)-1.0
def evalPolicy2(symbol, student_trades, startval, sd, ed, market_impact,commission_cost):
orders_df = pd.DataFrame(columns=['Shares','Order','Symbol'])
for row_idx in student_trades.index:
nshares = student_trades.loc[row_idx][0]
if nshares == 0:
continue
order = 'sell' if nshares < 0 else 'buy'
new_row = pd.DataFrame([[abs(nshares),order,symbol],],columns=['Shares','Order','Symbol'],index=[row_idx,])
orders_df = orders_df.append(new_row)
portvals = compute_portvals(orders_df, sd, ed, startval,market_impact,commission_cost)
return float(portvals[-1]/portvals[0])-1
def compute_portvals(orders_df, start_date, end_date, startval, market_impact=0.0, commission_cost=0.0):
"""Simulate the market for the given date range and orders file."""
symbols = []
orders = []
orders_df = orders_df.sort_index()
for date, order in orders_df.iterrows():
shares = order['Shares']
action = order['Order']
symbol = order['Symbol']
if action.lower() == 'sell':
shares *= -1
order = (date, symbol, shares)
orders.append(order)
symbols.append(symbol)
symbols = list(set(symbols))
dates = pd.date_range(start_date, end_date)
prices_all = util.get_data(symbols, dates)
prices = prices_all[symbols]
prices = prices.fillna(method='ffill').fillna(method='bfill')
prices['_CASH'] = 1.0
trades = pd.DataFrame(index=prices.index, columns=symbols)
trades = trades.fillna(0)
cash = pd.Series(index=prices.index)
cash = cash.fillna(0)
cash.iloc[0] = startval
for date, symbol, shares in orders:
price = prices[symbol][date]
val = shares * price
# transaction cost model
val += commission_cost + (pd.np.abs(shares)*price*market_impact)
positions = prices.loc[date] * trades.sum()
totalcash = cash.sum()
if (date < prices.index.min()) or (date > prices.index.max()):
continue
trades[symbol][date] += shares
cash[date] -= val
trades['_CASH'] = cash
holdings = trades.cumsum()
df_portvals = (prices * holdings).sum(axis=1)
return df_portvals
if __name__ == "__main__":
pytest.main(["-s", __file__])

9
strategy_evaluation/indicators.py
View File

@ -73,7 +73,7 @@ def rsi(df, symbol, period=14):
(avg_loss / period))))
return rsi
key = f"rsi"
key = "rsi"
# Add one to get 'period' price changes (first change is nan).
period += 1
df[key] = df[symbol].rolling(period).apply(rsi)
@ -91,13 +91,6 @@ def macd(df, symbol):
return df[[k1, k2, k3]]
def price_delta(df, symbol, period=1):
"""Calculate delta between previous day and today."""
k = f"diff_{period}"
df[k] = df[symbol].diff(periods=period)
return df[k]
def price_delta(df, symbol, period=1):
"""Calculate percentage change for period."""
k = f"pct_{period}"