145 lines
5.5 KiB
Python
145 lines
5.5 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
小唯股票批量回测 — 7只关注股票 MA20 策略对比
|
||
========================================
|
||
跑完全部关注股票的 MA20 回测，输出汇总对比表。
|
||
|
||
用法:
|
||
python3 stock_backtest_all.py # 批量回测全部关注股票
|
||
python3 stock_backtest_all.py --push # 同时推送飞书
|
||
python3 stock_backtest_all.py --json # 只输出 JSON 汇总
|
||
"""
|
||
|
||
import json, sys, urllib.request
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
OUTPUT = Path.home() / ".hermes" / "stock_backtest"
|
||
OUTPUT.mkdir(exist_ok=True)
|
||
FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/446db983-e392-4d2c-bfb8-f9060e5df3ad"
|
||
|
||
# 与 stock_portfolio.py 一致的关注股票
|
||
WATCHED_STOCKS = [
|
||
("000858", "五粮液", "白酒"),
|
||
("600519", "贵州茅台", "白酒"),
|
||
("000568", "泸州老窖", "白酒"),
|
||
("002304", "洋河股份", "白酒"),
|
||
("600036", "招商银行", "银行"),
|
||
("601318", "中国平安", "保险"),
|
||
("000001", "平安银行", "银行"),
|
||
]
|
||
|
||
|
||
def send_feishu(msg):
|
||
payload = json.dumps({"msg_type": "text", "content": {"text": msg}}).encode()
|
||
req = urllib.request.Request(FEISHU_WEBHOOK, data=payload,
|
||
headers={"Content-Type": "application/json"})
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=10):
|
||
pass
|
||
except Exception:
|
||
print("⚠️ 飞书推送失败")
|
||
|
||
|
||
def load_result(code):
|
||
"""读取已保存的回测结果"""
|
||
f = OUTPUT / f"ma20_result_{code}.json"
|
||
if f.exists():
|
||
try:
|
||
return json.load(open(f))
|
||
except Exception:
|
||
return None
|
||
return None
|
||
|
||
|
||
def run_all(push=False):
|
||
# 逐个调用回测脚本
|
||
import subprocess
|
||
for code, name, industry in WATCHED_STOCKS:
|
||
print(f"⏳ 回测 {name}({code}) ...")
|
||
r = subprocess.run(
|
||
[sys.executable, str(Path(__file__).parent / "stock_ma20_backtest.py"), code, name],
|
||
capture_output=True, text=True, timeout=120)
|
||
if r.returncode != 0:
|
||
print(f" ❌ {name} 回测失败: {r.stderr[-300:]}")
|
||
else:
|
||
print(f" ✅ {name} 完成")
|
||
|
||
# 汇总对比
|
||
rows = []
|
||
for code, name, industry in WATCHED_STOCKS:
|
||
d = load_result(code)
|
||
if d:
|
||
rows.append({
|
||
"code": code, "name": name, "industry": industry,
|
||
"strategy_return": d.get("strategy_return", 0),
|
||
"buyhold_return": d.get("buyhold_return", 0),
|
||
"alpha": d.get("alpha", 0),
|
||
"max_drawdown": d.get("max_drawdown", 0),
|
||
"total_trades": d.get("total_trades", 0),
|
||
"win_rate": d.get("win_rate", 0),
|
||
"in_position": d.get("in_position", False),
|
||
"current_price": d.get("current_price", 0),
|
||
})
|
||
|
||
rows.sort(key=lambda x: x["alpha"], reverse=True)
|
||
|
||
print(f"\n{'='*88}")
|
||
print(f"小唯股票 MA20 策略批量回测汇总 {datetime.now().strftime('%Y-%m-%d')}")
|
||
print(f"{'='*88}")
|
||
print(f"{'股票':<12} {'策略收益':>8} {'买入持有':>8} {'α超额':>7} {'最大回撤':>7} {'交易':>4} {'胜率':>6} 状态")
|
||
print("-" * 88)
|
||
for r in rows:
|
||
pos = "🟢持仓" if r["in_position"] else "⚪空仓"
|
||
print(f"{r['name']}({r['code']}) {r['strategy_return']:>+7.2f}% {r['buyhold_return']:>+7.2f}% "
|
||
f"{r['alpha']:>+6.2f}% {r['max_drawdown']:>6.2f}% {r['total_trades']:>4} {r['win_rate']:>5.1f}% {pos}")
|
||
print("-" * 88)
|
||
|
||
# 结论：策略适用边界
|
||
good = [r for r in rows if r["alpha"] > 0]
|
||
bad = [r for r in rows if r["alpha"] <= 0]
|
||
print(f"\n📌 策略适用性结论:")
|
||
print(f" ✅ α>0 (策略有效): {', '.join(r['name'] for r in good) if good else '无'}")
|
||
print(f" ⚠️ α≤0 (策略无效/跑输): {', '.join(r['name'] for r in bad) if bad else '无'}")
|
||
best = rows[0] if rows else None
|
||
if best:
|
||
print(f" 🏆 最佳标的: {best['name']} (α={best['alpha']:+.2f}%)")
|
||
print(f"\n⚠️ 提醒: MA20策略在单边下跌/长期阴跌股上会频繁止损（低胜率高赔率），"
|
||
f"选股比策略本身更重要。")
|
||
|
||
# 保存汇总
|
||
summary = {
|
||
"generated": datetime.now().strftime("%Y-%m-%d %H:%M"),
|
||
"strategy": "MA20突破",
|
||
"stocks": rows,
|
||
"effective": [r["name"] for r in good],
|
||
"ineffective": [r["name"] for r in bad],
|
||
"best": best["name"] if best else None,
|
||
}
|
||
summary_file = OUTPUT / "ma20_summary.json"
|
||
with open(summary_file, "w") as f:
|
||
json.dump(summary, f, ensure_ascii=False, indent=2)
|
||
print(f"\n📄 汇总已保存: {summary_file}")
|
||
|
||
# 飞书推送
|
||
if push:
|
||
lines = [f"📊 MA20策略批量回测 {datetime.now().strftime('%m-%d')}"]
|
||
for r in rows:
|
||
pos = "🟢" if r["in_position"] else "⚪"
|
||
lines.append(f"{pos} {r['name']}: 策略{r['strategy_return']:+.1f}% 持有{r['buyhold_return']:+.1f}% "
|
||
f"α{r['alpha']:+.1f}% 回撤{r['max_drawdown']:.0f}% 胜率{r['win_rate']:.0f}%")
|
||
lines.append(f"\n🏆 最佳: {best['name'] if best else '-'} | "
|
||
f"策略有效: {', '.join(r['name'] for r in good) if good else '无'}")
|
||
send_feishu("\n".join(lines))
|
||
|
||
return rows
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if "--json" in sys.argv:
|
||
import json as _json
|
||
rows = run_all(push=False)
|
||
print(_json.dumps(rows, ensure_ascii=False, indent=2))
|
||
else:
|
||
run_all(push="--push" in sys.argv)
|