问题描述
我有一个看起来像这样的 csv 文件/pandas dataframe。它包含根据我自己的计算每天重新平衡的投资组合的各种投资组合组合。
date asset percentage
4-Jan-21 AAPL 12.00%
4-Jan-21 TSM 1.00%
4-Jan-21 IBM 31.00%
4-Jan-21 KO 15.00%
4-Jan-21 AMD 41.00%
5-Jan-21 DELL 23.00%
5-Jan-21 TSM 12.20%
5-Jan-21 IBM 15.24%
5-Jan-21 KO 1.50%
5-Jan-21 NKE 7.50%
5-Jan-21 TSLA 9.50%
5-Jan-21 CSCO 3.30%
5-Jan-21 JPM 27.76%
6-Jan-21 AMD 45%
6-Jan-21 BA 0.50%
6-Jan-21 ORCL 54.50%
7-Jan-21 AAPL 50.00%
7-Jan-21 KO 50.00%
...
我想测试一个包含 12 个资产组合的策略。
AAPL,TSM,IBM,KO,AMD,DELL,NKE,TSLA,CSCO,JPM,BA,ORCL
假设在 2021 年 1 月 4 日，投资组合的组成将是苹果的 12%，TSM 的 1%……等等。我希望能够查看价格并知道我应该持有多少。
第二天，即 2021 年 1 月 5 日，戴尔的构成将变为 23%。等，如果该股票不在此列表中，则意味着当天为 0%。
我一直将 backTrader 视为回测平台，但是，我在 repo 中看到的代码主要展示了如何使用指标进行操作，例如 SMA 交叉、RSI...
我的问题是：是否可以根据我拥有的这些组合创建和测试投资组合，以便检查该策略的回报？它会检查这个框架，并知道在特定日期买入或卖出的股票代码中有多少股票。
所以我买入或卖出的股票范围是 AAPL,ORCL
所以在 1 月 4 日 21 日，它可能看起来像，
dictionary['4Jan2021'] = {'AAPL':0.12,'TSM':0.01,'IBM':0.31,'KO':0.15,'AMD':0.41,}
1 月 5 日 21 日，它看起来像，
dictionary['5Jan2021'] = {'DELL':0.23,'TSM':0.122,'IBM':0.1524,'KO':0.015,'NKE':0.075,'TSLA':0.095,'CSCO':0.033,'JPM':0.2776,}
如果代码不存在，则表示其为 0%。 投资组合的构成需要每天都在变化。
解决方法
你要做的第一件事就是用你的数据加载你的目标。我喜欢 当我将它添加到 backtrader 时，我亲自将目标附加到数据线上。
tickers = {"FB": 0.25,"MSFT": 0.4,"TSLA": 0.35}
for ticker,target in tickers.items():
data = bt.feeds.YahooFinanceData(
dataname=ticker,timeframe=bt.TimeFrame.Days,fromdate=datetime.datetime(2019,1,1),todate=datetime.datetime(2020,12,31),reverse=False,)
data.target = target
cerebro.adddata(data,name=ticker)
接下来您将要查看每个数据，并确定当前分配。如果当前分配与所需分配（阈值）相差太远，则您交易所有数据。
注意有一个缓冲区变量。这将减少用于计算交易单位的账户的总价值。这有助于避免保证金。
您将使用字典来跟踪此信息。
def next(self):
track_trades = dict()
total_value = self.broker.get_value() * (1 - self.p.buffer)
for d in self.datas:
track_trades[d] = dict()
value = self.broker.get_value(datas=[d])
allocation = value / total_value
units_to_trade = (d.target - allocation) * total_value / d.close[0]
track_trades[d]["units"] = units_to_trade
# Can check to make sure there is enough distance away from ideal to trade.
track_trades[d]["threshold"] = abs(d.target - allocation) > self.p.threshold
检查所有阈值以确定是否交易。如果任何数据需要交易，那么所有数据都需要交易。
rebalance = False
for values in track_trades.values():
if values['threshold']:
rebalance = True
if not rebalance:
return
最后，执行您的交易。始终先出售以在帐户中产生现金并避免利润。
# Sell shares first
for d,value in track_trades.items():
if value["units"] < 0:
self.sell(d,size=value["units"])
# Buy shares second
for d,value in track_trades.items():
if value["units"] > 0:
self.buy(d,size=value["units"])
这里是所有代码供您参考。
import datetime
import backtrader as bt
class Strategy(bt.Strategy):
params = (
("buffer",0.05),("threshold",0.025),)
def log(self,txt,dt=None):
""" Logging function fot this strategy"""
dt = dt or self.data.datetime[0]
if isinstance(dt,float):
dt = bt.num2date(dt)
print("%s,%s" % (dt.date(),txt))
def print_signal(self):
self.log(
f"o {self.datas[0].open[0]:7.2f} "
f"h {self.datas[0].high[0]:7.2f} "
f"l {self.datas[0].low[0]:7.2f} "
f"c {self.datas[0].close[0]:7.2f} "
f"v {self.datas[0].volume[0]:7.0f} "
)
def notify_order(self,order):
""" Triggered upon changes to orders. """
# Suppress notification if it is just a submitted order.
if order.status == order.Submitted:
return
# Print out the date,security name,order number and status.
type = "Buy" if order.isbuy() else "Sell"
self.log(
f"{order.data._name:<6} Order: {order.ref:3d} "
f"Type: {type:<5}\tStatus"
f" {order.getstatusname():<8} \t"
f"Size: {order.created.size:9.4f} Price: {order.created.price:9.4f} "
f"Position: {self.getposition(order.data).size:5.2f}"
)
if order.status == order.Margin:
return
# Check if an order has been completed
if order.status in [order.Completed]:
self.log(
f"{order.data._name:<6} {('BUY' if order.isbuy() else 'SELL'):<5} "
# f"EXECUTED for: {dn} "
f"Price: {order.executed.price:6.2f} "
f"Cost: {order.executed.value:6.2f} "
f"Comm: {order.executed.comm:4.2f} "
f"Size: {order.created.size:9.4f} "
)
def notify_trade(self,trade):
"""Provides notification of closed trades."""
if trade.isclosed:
self.log(
"{} Closed: PnL Gross {},Net {},".format(
trade.data._name,round(trade.pnl,2),round(trade.pnlcomm,)
)
def next(self):
track_trades = dict()
total_value = self.broker.get_value() * (1 - self.p.buffer)
for d in self.datas:
track_trades[d] = dict()
value = self.broker.get_value(datas=[d])
allocation = value / total_value
units_to_trade = (d.target - allocation) * total_value / d.close[0]
track_trades[d]["units"] = units_to_trade
# Can check to make sure there is enough distance away from ideal to trade.
track_trades[d]["threshold"] = abs(d.target - allocation) > self.p.threshold
rebalance = False
for values in track_trades.values():
if values['threshold']:
rebalance = True
if not rebalance:
return
# Sell shares first
for d,value in track_trades.items():
if value["units"] < 0:
self.sell(d,size=value["units"])
# Buy shares second
for d,value in track_trades.items():
if value["units"] > 0:
self.buy(d,size=value["units"])
if __name__ == "__main__":
cerebro = bt.Cerebro()
tickers = {"FB": 0.25,"TSLA": 0.35}
for ticker,target in tickers.items():
data = bt.feeds.YahooFinanceData(
dataname=ticker,)
data.target = target
cerebro.adddata(data,name=ticker)
cerebro.addstrategy(Strategy)
# Execute
cerebro.run()
##################################
############# 编辑##############
##################################
每天增加每个证券的可变分配还有一个额外的要求。下面的代码实现了这一点。
import datetime
import backtrader as bt
class Strategy(bt.Strategy):
params = (
("buffer",)
)
def __init__(self):
for d in self.datas:
d.target = {
datetime.datetime.strptime(date,"%d-%b-%y").date(): allocation
for date,allocation in d.target.items()
}
def next(self):
date = self.data.datetime.date()
track_trades = dict()
total_value = self.broker.get_value() * (1 - self.p.buffer)
for d in self.datas:
if date not in d.target:
if self.getposition(d):
self.close(d)
continue
target_allocation = d.target[date]
track_trades[d] = dict()
value = self.broker.get_value(datas=[d])
current_allocation = value / total_value
net_allocation = target_allocation - current_allocation
units_to_trade = (
(net_allocation) * total_value / d.close[0]
)
track_trades[d]["units"] = units_to_trade
# Can check to make sure there is enough distance away from ideal to trade.
track_trades[d]["threshold"] = abs(net_allocation) > self.p.threshold
rebalance = False
for values in track_trades.values():
if values["threshold"]:
rebalance = True
if not rebalance:
return
# Sell shares first
for d,size=value["units"])
if __name__ == "__main__":
cerebro = bt.Cerebro()
allocations = [
("AAPL","4-Jan-21",0.300),("TSM",0.200),("IBM",("KO",0.2000),("AMD",0.1000),("DELL","5-Jan-21",0.20),0.1),("NKE",0.15),("TSLA",0.10),("CSCO",0.050),("JPM","6-Jan-21",0.25),("BA",("ORCL",0.50),("AAPL","7-Jan-21",0.5000),]
ticker_names = list(set([alls[0] for alls in allocations]))
targets = {ticker: {} for ticker in ticker_names}
for all in allocations:
targets[all[0]].update({all[1]: all[2]})
for ticker,target in targets.items():
data = bt.feeds.YahooFinanceData(
dataname=ticker,fromdate=datetime.datetime(2020,21),todate=datetime.datetime(2021,8),name=ticker)
cerebro.addstrategy(Strategy)
cerebro.broker.setcash(1000000)
# Execute
cerebro.run()