strategy · 四大库合体 2026-09-28

from datetime import datetime # datetime：Python 自带的时间模块 import numpy as np # numpy：数值计算（发动机） import pandas as pd # pandas：表格处理（工作台） from scipy import stats # scipy.stats：统计工具（工具箱） from quant_data import get_kline # 从数据工具箱拿"拉数据"的函数

#函数前置 def backtest_signals(df, fast=5, slow=20): """ 在表格上算出金叉/死叉信号，返回加了信号列的表格。 fast 是快速均线天数（默认 5 天），slow 是慢速均线天数（默认 20 天）。 """ close = df["close"] # 先取收盘价这一列 # 新增两列：5 日均线 和 20 日均线（rolling 滚动窗口均值） df["MA" + str(fast)] = close.rolling(fast).mean() df["MA" + str(slow)] = close.rolling(slow).mean() # "MA" + str(fast)：字符串拼接，5 和 20 是函数参数传进来的数字 # 所以列名会自动变成 MA5 和 MA20，想换成 10/60 也行 # numpy 的 np.where：满足条件的位置填 1，不满足填 0 df["signal"] = np.where(df["MA" + str(fast)] > df["MA" + str(slow)], 1, 0) # 1 表示：短线在长线上方（看多，持仓）；0 表示：下方（看空，空仓） # 找"信号变动"的那一天：0->1 是金叉(买入)，1->0 是死叉(卖出) df["signal_change"] = np.diff(df["signal"], prepend=0) # np.diff：算前后两天的差值； # prepend=0：在最前面补一个 0 让长度对齐； # 防"未来函数"：信号要"明天才生效"，不能当天看信号当天买 # .shift(1)：把整列往下挪一行；.fillna(0)：挪出来的空位填 0 df["position"] = df["signal"].shift(1).fillna(0) return df # 返回加了新列的表格

if __name__ == "__main__": # ========== ① datetime 定日期 + 拉数据 ========== # end_date 用今天的日期动态生成 df = get_kline(start_date="20260101", end_date=datetime.now().strftime("%Y%m%d"))

# ========== ② pandas 管表 ========== df["收益率"] = df["close"].pct_change() # 新增一列：每日收益率

# ========== ③ numpy 算信号 ========== df = backtest_signals(df) # 调用上面封装的函数，算出金叉死叉信号

# ========== ④ scipy 评估 ========== # 策略收益 = 持仓状态(0/1) × 当日收益率（持仓才赚钱，空仓不亏不赚） df["策略收益"] = df["position"] * df["收益率"] sr = df["策略收益"].dropna() # 去掉空值 # 策略夏普比率 sharpe = (sr.mean() * 252) / (sr.std() * np.sqrt(252)) if sr.std() else 0.0 t_stat, p_value = stats.ttest_1samp(sr, 0) # t 检验：策略收益是不是显著

print("\n===== 最近 5 次买卖信号 =====") trades = df[df["signal_change"] != 0] # 只留"信号变动"的日子 if len(trades) > 0: # 条件分支：确实有信号 recent = trades.tail(5).copy() # 取最近 5 条；.copy() 复制一份防止改坏原表 action_list = [] # 建一个空列表，用来装"动作"文字 for v in recent["signal_change"]: # 循环：一行一行看信号变动值 if v == 1: # 0 变 1 action_list.append("买入(金叉)") # 装"买入" elif v == -1: # 1 变 0 action_list.append("卖出(死叉)") # 装"卖出" else: # 其他情况 action_list.append("无") # 装"无" recent["动作"] = action_list # 把动作列表挂成新的一列 print(recent[["动作", "close"]].to_string()) # 打印动作和当日收盘价 last = trades.iloc[-1] # .iloc：按位置取行；[-1] 是最后一行 if last["signal_change"] == 1: # 判断最后一次是买还是卖 last_action = "买入(金叉)" else: last_action = "卖出(死叉)" # last.name 就是最后一行的时间（因为日期是索引） print(f"\n最近一次信号：{last.name.date()} {last_action}，当日收盘 {last['close']:.2f}") else: # 没有信号的情况 print("样本期内没有出现完整信号") print("\n===== 策略评估 =====") print(f"样本期策略平均日收益 : {sr.mean() * 100:.4f}%") # 平均日收益 print(f"策略夏普比率 : {sharpe:.2f}") print(f"t 检验 p 值 : {p_value:.4f} （<0.05 表示显著）") print("\n⚠️ 金叉死叉只是最简单的教学思路，本脚本不构成投资建议")