本期将从零开始，手动搭建一个基于规则的简单量化交易系统，旨在提高对整个过程的理解能力。
Step 1: 数据加载
其中股票数据采用已存好的CSV数据，数据的获取方式可参见
另外采用 TA-lib 的 RSI 技术指标构建买入卖出信号，详情可参见
# Step 1: load dataset and generate features
def prepare_data(codes=['000300.SH', '399006.SZ'], start_time="20100101", end_time="20211231"):
df = load_data(codes, start_time, end_time)
df["rsi"] = ta.RSI(df.close, timeperiod=14)
df["to_buy"] = ""
df.loc[df[&#34;rsi&#34;] <= 30, &#39;to_buy&#39;] = True
df[&#39;to_buy&#39;] = df[&#39;to_buy&#39;].astype(&#39;bool&#39;)
df[&#34;to_sell&#34;] = &#34;&#34;
df.loc[df[&#34;rsi&#34;] >= 70, &#39;to_sell&#39;] = True
df[&#39;to_sell&#39;] = df[&#39;to_sell&#39;].astype(&#39;bool&#39;)
return dfStep 2: 指定策略
策略规则非常简单：
- 账户初始值100000，每只股票占相同比重
- 根据数据的买入信号买入，卖出信号卖出
# Step 2: prepare strategy
class SelectBySignal(object):
def __init__(self, signal_buy=&#39;to_buy&#39;, signal_sell=&#39;to_sell&#39;):
super(SelectBySignal, self).__init__()
self.signal_buy = signal_buy
self.signal_sell = signal_sell
def __call__(self, context):
bar = context[&#39;bar&#39;].copy()
acc = context[&#39;acc&#39;]
holding = acc.get_holding_instruments()
to_buy = list(bar[bar[self.signal_buy]].index)
to_sell = list(bar[bar[self.signal_sell]].index)
instruments = to_buy + holding
to_selected = []
for s in instruments:
if s not in to_sell:
to_selected.append(s)
context[&#39;selected&#39;] = to_selected
n = len(to_selected)
if n > 0:
context[&#39;weights&#39;] = {code:1/n for code in to_selected}
else:
context[&#39;weights&#39;] = {}
return False
class Strategy:
def __init__(self, algo=None):
self.algo = algo
self.acc = Account()
def algo_processor(self, context):
if self.algo(context) is True: #如果algo返回True,直接不运行，本次不调仓
return None
return context[&#39;weights&#39;]
def onbar(self, index, date, df_bar):
self.acc.update_bar(date, df_bar)
weights = self.algo_processor({&#39;index&#39;: index, &#39;bar&#39;:df_bar, &#39;date&#39;:date, &#39;acc&#39;:self.acc})
if weights is not None:
self.acc.adjust_weights(date, weights)Step 3: 回测
根据上述数据和策略进行回测
# Step 3: backtest
class Backtest:
def __init__(self, df):
self.df = df
self.dates = self.df.index.unique()
self.observers = []
def onbar(self, index, date):
df_bar = self.df.loc[date]
if type(df_bar) is pd.Series:
df_bar = df_bar.to_frame().T
df_bar.index = df_bar[&#39;code&#39;]
self.strategy.onbar(index, date, df_bar)
def run(self, s):
self.strategy = s
for index, date in enumerate(self.dates):
self.onbar(index, date)
return self.get_results()
def get_results(self):
s = self.strategy
df = s.acc.get_results_df()
return dfStep 4: 分析
分析该策略的表现，并与沪深300指数进行对比
# Step 4: analysis
def analysis(start, end, benchmarks=[]):
equities = []
for benchmark in benchmarks:
bench_df = load_from_file(benchmark)[start:end]
se = (bench_df[&#39;rate&#39;] + 1).cumprod()
se.name = benchmark
equities.append(se)
path = os.path.dirname(__file__)
filename = os.path.dirname(path)+&#39;/results/first_test.csv&#39;
if os.path.exists(filename):
df = pd.read_csv(filename)
df[&#39;date&#39;] = df[&#39;date&#39;].apply(lambda x: str(x))
df.index = df[&#39;date&#39;]
se = (df[&#39;rate&#39;] + 1).cumprod()
se.name = &#39;strategy&#39;
equities.append(se)
df_equities = pd.concat(equities, axis=1)
df_equities.dropna(inplace=True)
print(df_equities)
from performance import PerformanceUtils
df_ratios, df_corr, df_years = PerformanceUtils().calc_equity(df_equity=df_equities)
return df_equities, df_ratios, df_corr, df_years运行主函数
if __name__ == &#39;__main__&#39;:
date_start = &#34;20100101&#34;
date_end = &#34;20211231&#34;
df = prepare_data(codes=[&#39;000300.SH&#39;, &#39;399006.SZ&#39;], start_time=date_start, end_time=date_end)
algo = SelectBySignal(signal_buy=&#39;to_buy&#39;, signal_sell=&#39;to_sell&#39;)
s = Strategy(algo=algo)
b = Backtest(df=df)
df = b.run(s)
path = os.path.dirname(__file__)
df.to_csv(os.path.dirname(path) + &#39;/results/first_test.csv&#39;)
df_equities, df_ratios, df_corr, df_years = analysis(start=date_start, end=date_end, benchmarks=[&#39;000300.SH&#39;])
display(df_ratios)
fig = plt.figure(figsize=(8, 6))
ax1 = fig.add_subplot(2, 1, 1)
ax2 = fig.add_subplot(2, 1, 2)
df_equities.plot(ax=ax1)
if df_years is not None:
print(df_years)
df_years.T.plot(kind=&#39;bar&#39;, ax=ax2, use_index=True)
plt.show()得到结果如下：
分别为累积的各项数据对比，以及各年的年化对比
![]()
![]() ：策略最终股票和现金的总价值
![]() ：策略开始股票和现金的总价值
![]()
![]() ：策略收益
![]() ：策略执行天数
- 波动率（Algorithm Volatility）
用来测量策略的风险性，波动越大代表策略风险越高
![]()
![]() ：策略每日收益率
![]() ：策略每日收益率的平均值
表示每承受一单位总风险，会产生多少的超额报酬，可以同时对策略的收益与风险进行综合考虑。
![]()
![]() ：策略年化收益率
![]() ：无风险利率（默认0.04）
![]() ：策略收益波动率
累计收益 年化收益 波动率 夏普比 最大回撤 年化收益与最大回撤比
000300.SH 0.413 0.030 0.228 0.132 -0.467 0.064
strategy 1.241 0.073 0.248 0.294 -0.576 0.127
2010 2011 2012 2013 2014 ... 2017 2018 2019 2020 2021
000300.SH -0.117 -0.265 0.098 -0.077 0.522 ... 0.206 -0.263 0.380 0.255 -0.044
strategy -0.051 -0.316 0.057 0.319 0.305 ... 0.040 -0.276 0.424 0.430 0.044做成图如下所示：
![]()
完整代码参见
另外也可参考： |
|
转播
