# base.py
def compose_signal(data):
# 整合交易信号,连续出现两次买入信号只保留第一次
data['buy_signal'] = np.where((data['buy_signal'] == 1) & (data['buy_signal'].shift(1) == 1), 0, data['buy_signal'])
data['sell_signal'] = np.where((data['sell_signal'] == -1) & (data['sell_signal'].shift(1) == -1), 0,
data['sell_signal'])
# 合并信号,1买入,-1卖出,后面按照这个来计算收益
data['signal'] = data['buy_signal'] + data['sell_signal']
return data
def calculate_prof_pct(data):
# 计算单次收益率:开仓、平仓(开仓的全部股数)
data.loc[data['signal'] != 0, 'profit_pct'] = data['close'].pct_change()
data = data[data['signal'] == -1] # 筛选平仓后的数据:单次收益
return data
def calculate_cum_prof(data):
df = data.copy()
df['cum_profit'] = pd.DataFrame(1+ df['profit_pct']).cumprod()-1
return df
# ma.py
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import data.stock as st
import strategy.base as stra
def ma_strategy(data, short_windows=5, long_window=20):
# 计算技术指标,ma短期,ma长期
data['short_ma'] = data['close'].rolling(window=short_windows).mean()
data['long_ma'] = data['close'].rolling(window=long_window).mean()
# 生成信号:金叉买入,死叉卖出
data['buy_signal'] = np.where(data['short_ma'] > data['long_ma'], 1, 0)
data['sell_signal'] = np.where(data['short_ma'] < data['long_ma'], -1, 0)
# 过滤信号
data = stra.compose_signal(data)
# 计算单次收益
data = stra.calculate_prof_pct(data)
# 计算累计收益
data = stra.calculate_cum_prof(data)
return data
if __name__ == '__main__':
# # 计算多个股的累计收益率
stocks = ['000001.XSHE', '000858.XSHE', '002594.XSHE']
cum_profits = pd.DataFrame()
for code in stocks:
df = st.get_single_price(code=code, time_freq='daily', start_date='2016-01-01', end_date='2021-01-01')
df = ma_strategy(df)
cum_profits[code] = df['cum_profit'].reset_index(drop=True)
df['cum_profit'].plot(label=code)
print(cum_profits)
plt.legend()
plt.show()
说明:为了避免 SettingwithCopyWarning 的问题,我改写了 calculate_cum_prof 函数,即便是按照视频 calculate_cum_prof 函数写法,我测试结果依旧一样,和视频结果大相径庭。不知道哪里有问题,我核对了所有函数,没发现哪里有写错。
另:4-11 寻找最优参数章节计算的数字和视频吻合,说明 ma_strategy 函数无误,我裂了…