From cfc14d15d7e72a8b39017abbeeb65d8d7dff9e9e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=B0=8F=E5=94=AF=20A06?= Date: Sun, 12 Jul 2026 00:50:59 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=87=AA=E9=80=82=E5=BA=94=E7=AD=96?= =?UTF-8?q?=E7=95=A5=E7=B3=BB=E7=BB=9F=20+=205=E5=8F=AA=E8=82=A1=E7=A5=A8?= =?UTF-8?q?=E5=AE=8C=E6=95=B4=E6=B5=8B=E8=AF=95=E7=BB=93=E8=AE=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - stock_adaptive.py: ADX环境感知策略切换(MA20/MACD) - 5只股票批量测试结果: - 茅台+五粮液(下跌): α=+1.1%/+39.2% ✅ 策略有效 - 平安银行(震荡): α=-3.8% - 宁德/沪深300(强势): α=-28.9%/-9.8% ❌ 策略有害 - 核心结论: 趋势明确时少动，趋势混乱时用MA20突破 - cangjie-skills INDEX.md更新测试结论 --- cangjie-skills/股票投研体系/INDEX.md | 31 ++- scripts/stock_adaptive.py | 341 +++++++++++++++++++++++++++ 2 files changed, 367 insertions(+), 5 deletions(-) create mode 100644 scripts/stock_adaptive.py diff --git a/cangjie-skills/股票投研体系/INDEX.md b/cangjie-skills/股票投研体系/INDEX.md index 9baedf23..4fc397cf 100644 --- a/cangjie-skills/股票投研体系/INDEX.md +++ b/cangjie-skills/股票投研体系/INDEX.md @@ -4,7 +4,7 @@ 更新: 2026-07-12 策略模拟结论 状态: Phase5 模拟验证进行中 -## 策略模拟结论（2026-07-12） +## 自适应策略模拟结论（2026-07-12） 贵州茅台(600519) 2023-01 ~ 2026-07，买入持有基准: -13.8% @@ -17,6 +17,26 @@ **结论**: MA20日突破策略最优，减少亏损+跑赢大盘 **注意**: 贵州茅台整体下跌，所有策略均亏损，策略价值在于减少损失 +--- + +## 自适应策略测试结果（2026-07-12） + +ADX>25用MA20突破，ADX≤25用MACD。测5只股票: + +| 股票 | 基准收益 | 自适应α | 评价 | +|------|---------|---------|------| +| 贵州茅台 | -13.8% | +1.1% ✅ | 下跌市有效 | +| 五粮液 | -39.2% | +39.2% ✅ | 大幅跑赢 | +| 平安银行 | +16.8% | -3.8% | 震荡市略输 | +| 宁德时代 | +98.9% | -28.9% ❌ | 强势股策略干扰 | +| 沪深300ETF | +36.2% | -9.8% ❌ | 强势ETF策略干扰 | +| **平均** | - | **-0.5%** | 基本持平 | + +**关键规律**: +- 策略在**下跌趋势**中有效（茅台、五粮液） +- 策略在**强势趋势**中有害（宁德+99%→策略只赚了70%） +- **通用结论**: 趋势明确时少动，趋势混乱时用策略 + ## 已蒸馏 Skill | # | 文件 | 主题 | 核心要点 | @@ -33,10 +53,11 @@ Phase 5: 模拟交易与策略验证 - ✅ 数据源: 腾讯/ifzq K线API (前复权日K) -- ✅ 回测引擎: 纯Python (stock_compare.py) -- ✅ 策略对比: MACD / MA突破 / 双均线 -- ⬜ 选股范围: 扩大多只股票测试 -- ⬜ 实盘模拟: 达到预期后小仓位实操 +- ✅ 回测引擎: 纯Python (stock_compare.py / stock_adaptive.py) +- ✅ 策略对比: MACD / MA突破 / 双均线 / 自适应 +- ✅ 5只股票测试完成（茅台/五粮液/平安银行/宁德时代/沪深300ETF） +- ⬜ 选股策略: 选弱势/震荡股，策略有效 +- ⬜ 小仓位实盘模拟: 选1-2只股票跑策略，验证真实效果 ## 方法论来源 diff --git a/scripts/stock_adaptive.py b/scripts/stock_adaptive.py new file mode 100644 index 00000000..f0e55436 --- /dev/null +++ b/scripts/stock_adaptive.py @@ -0,0 +1,341 @@ +#!/usr/bin/env python3 +""" +小唯自适应策略 — 智能切换MACD/MA20/买入持有 +========================================== +根据市场环境自动选择最优策略 + +判断逻辑: +- 趋势强度(ADX) > 25 → 趋势市场 → MA20突破(追涨) +- ADX <= 25 → 震荡市场 → MACD(区间波动) +- 持仓时ADX突然下降 → 快速离场 + +用法: + python3 stock_adaptive.py <代码> [起始] [结束] + python3 stock_adaptive.py --batch # 批量测5只股票 +""" + +import json, sys, urllib.request +from datetime import datetime +from pathlib import Path + +import numpy as np +import pandas as pd + +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" + +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: + pass + + +def get_data(code, start, end): + mc = f"sh{code}" if code.startswith("6") else f"sz{code}" + url = (f"https://web.ifzq.gtimg.cn/appstock/app/fqkline/get" + f"?_var=kline_dayqfq¶m={mc},day,{start},{end},500,qfq") + try: + text = urllib.request.urlopen(url, timeout=10).read().decode("utf-8") + data = json.loads(text.replace("kline_dayqfq=", "", 1)) + qfq = (data.get("data", {}).get(mc, {}).get("qfqday") or + data.get("data", {}).get(mc, {}).get("day") or []) + rows = [] + for item in qfq: + if len(item) < 6: + continue + try: + rows.append({"date": item[0], "open": float(item[1]), + "close": float(item[2]), "high": float(item[3]), + "low": float(item[4]), "volume": float(item[5])}) + except (ValueError, IndexError): + continue + df = pd.DataFrame(rows) + df["date"] = pd.to_datetime(df["date"]) + df.set_index("date", inplace=True) + df.sort_index(inplace=True) + return df + except Exception: + return None + + +def compute_adx(df, n=14): + """计算ADX指标(趋势强度)，>25表示趋势市场""" + high = df["high"] + low = df["low"] + close = df["close"] + + # +DM, -DM + plus_dm = high.diff() + minus_dm = -low.diff() + plus_dm[plus_dm < 0] = 0 + minus_dm[minus_dm < 0] = 0 + + # True Range + tr1 = high - low + tr2 = abs(high - close.shift()) + tr3 = abs(low - close.shift()) + tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1) + + # ADX + atr = tr.rolling(n).mean() + plus_di = (plus_dm.rolling(n).mean() / atr * 100).fillna(0) + minus_di = (minus_dm.rolling(n).mean() / atr * 100).fillna(0) + + dx = (abs(plus_di - minus_di) / (plus_di + minus_di + 1e-9) * 100).fillna(0) + adx = dx.rolling(n).mean() + return adx + + +def adaptive_backtest(df, cash=100000): + """自适应策略: ADX>25用MA20，ADX<=25用MACD""" + closes = df["close"].values + n = len(closes) + + # 预计算所有指标 + ma20 = pd.Series(closes).rolling(20).mean() + adx = compute_adx(df, 14) + adx_vals = adx.fillna(0).values + + ema12 = pd.Series(closes).ewm(span=12).mean() + ema26 = pd.Series(closes).ewm(span=26).mean() + macd = ema12 - ema26 + signal = macd.ewm(span=9).mean() + macd_vals = macd.values + signal_vals = signal.values + + # 持仓状态 + shares = 0 + c = cash + peak = cash + max_dd = 0.0 + trades = [] + regime_changes = [] # 记录策略切换 + + for i in range(26, n): + price = closes[i] + regime = "TREND" if adx_vals[i] > 25 else "RANGE" + + # === MA20突破信号(趋势市场) === + if regime == "TREND": + if shares == 0: + if closes[i] > ma20.iloc[i] and closes[i-1] <= ma20.iloc[i-1]: + s = int(c / price) + if s > 0: + c -= s * price + shares = s + trades.append(("BUY", df.index[i], s, price, "MA20_TREND")) + elif shares > 0: + if closes[i] < ma20.iloc[i] and closes[i-1] >= ma20.iloc[i-1]: + c += shares * price + trades.append(("SELL", df.index[i], shares, price, "MA20_TREND")) + shares = 0 + + # === MACD信号(震荡市场) === + else: # RANGE + if shares == 0: + if macd_vals[i] > signal_vals[i] and macd_vals[i-1] <= signal_vals[i-1]: + s = int(c / price) + if s > 0: + c -= s * price + shares = s + trades.append(("BUY", df.index[i], s, price, "MACD_RANGE")) + elif shares > 0: + if macd_vals[i] < signal_vals[i] and macd_vals[i-1] >= signal_vals[i-1]: + c += shares * price + trades.append(("SELL", df.index[i], shares, price, "MACD_RANGE")) + shares = 0 + + equity = c + shares * price + peak = max(peak, equity) + dd = (peak - equity) / peak * 100 if peak > 0 else 0 + max_dd = max(max_dd, dd) + + final = c + shares * closes[-1] + ret = (final - cash) / cash * 100 + bh = (closes[-1] - closes[0]) / closes[0] * 100 + + # 统计 + total_trades = len([t for t in trades if t[0] == "SELL"]) + ma20_trades = len([t for t in trades if t[0] == "SELL" and t[4] == "MA20_TREND"]) + macd_trades = len([t for t in trades if t[0] == "SELL" and t[4] == "MACD_RANGE"]) + wins = len([t for t in trades if t[0] == "SELL" and t[3] > 0]) + + return { + "final": final, "return": ret, "buyhold": bh, + "alpha": ret - bh, "max_dd": max_dd, + "total_trades": total_trades, + "ma20_trades": ma20_trades, + "macd_trades": macd_trades, + "winrate": wins / total_trades * 100 if total_trades > 0 else 0, + "trades": [(t[0], str(t[1].date()), t[2], t[3], t[4]) for t in trades], + } + + +def run_benchmark(df, cash=100000): + """跑三个基准策略用于对比""" + closes = df["close"].values + + # MACD + ema12 = pd.Series(closes).ewm(span=12).mean() + ema26 = pd.Series(closes).ewm(span=26).mean() + macd = ema12 - ema26 + signal = macd.ewm(span=9).mean() + shares = 0; c = cash; trades = 0 + for i in range(26, len(closes)): + p = closes[i] + if macd.iloc[i] > signal.iloc[i] and macd.iloc[i-1] <= signal.iloc[i-1]: + if shares == 0: + n = int(c / p); c -= n * p; shares = n + elif macd.iloc[i] < signal.iloc[i] and macd.iloc[i-1] >= signal.iloc[i-1]: + if shares > 0: + c += shares * p; shares = 0; trades += 1 + final = c + shares * closes[-1] + macd_ret = (final - cash) / cash * 100 + + # MA20 + ma20 = pd.Series(closes).rolling(20).mean() + shares = 0; c = cash; trades = 0 + for i in range(20, len(closes)): + p = closes[i] + if closes[i] > ma20.iloc[i] and closes[i-1] <= ma20.iloc[i-1]: + if shares == 0: + n = int(c / p); c -= n * p; shares = n + elif closes[i] < ma20.iloc[i] and closes[i-1] >= ma20.iloc[i-1]: + if shares > 0: + c += shares * p; shares = 0; trades += 1 + final = c + shares * closes[-1] + ma20_ret = (final - cash) / cash * 100 + + bh = (closes[-1] - closes[0]) / closes[0] * 100 + return {"macd": macd_ret, "ma20": ma20_ret, "buyhold": bh} + + +def analyze(code, start, end=None): + end = end or datetime.now().strftime("%Y-%m-%d") + df = get_data(code, start, end) + if df is None or len(df) < 60: + print(f"数据获取失败"); return + + result = adaptive_backtest(df) + bench = run_benchmark(df) + + print(f"\n{'='*60}") + print(f"自适应策略 vs 基准 | {code} | {start} ~ {end}") + print(f"{'='*60}") + print(f"{'策略':<22} {'收益':>10} {'α':>10} {'最大回撤':>10} {'交易':>6}") + print(f"{'-'*60}") + print(f"{'自适应(ADX切换)':<22} {result['return']:>+9.1f}% {result['alpha']:>+9.1f}% {result['max_dd']:>9.1f}% {result['total_trades']:>6}") + print(f"{'MA20突破(固定)':<22} {bench['ma20']:>+9.1f}% {bench['ma20']-bench['buyhold']:>+9.1f}%") + print(f"{'MACD(固定)':<22} {bench['macd']:>+9.1f}% {bench['macd']-bench['buyhold']:>+9.1f}%") + print(f"{'买入持有':<22} {bench['buyhold']:>+9.1f}% {'基准':>9}") + print(f"{'='*60}") + print(f"自适应内: MA20触发{result['ma20_trades']}次, MACD触发{result['macd_trades']}次") + print(f"胜率: {result['winrate']:.0f}%") + + # 飞书 + msg = f"""📊 自适应策略对比 + +代码: {code} | {start} ~ {end} | {len(df)}条 + +策略收益对比: +自适应(ADX切换): {result['return']:+.1f}% (α={result['alpha']:+.1f}%, 回撤{result['max_dd']:.1f}%, {result['total_trades']}笔) +MA20突破(固定): {bench['ma20']:+.1f}% +MACD(固定): {bench['macd']:+.1f}% +买入持有基准: {bench['buyhold']:+.1f}% + +自适应内: MA20触发{result['ma20_trades']}次, MACD触发{result['macd_trades']}次 +胜率: {result['winrate']:.0f}% + +生成: {datetime.now().strftime('%Y-%m-%d %H:%M')} +小唯股票投研 · 自适应策略""" + send_feishu(msg) + + result_file = OUTPUT / f"adaptive_{code}.json" + with open(result_file, "w") as f: + json.dump({"code": code, "start": start, "end": end, + "adaptive": result, "benchmark": bench}, f, ensure_ascii=False, indent=2, default=str) + print(f"数据存: {result_file}") + return result + + +def batch_test(): + """批量测5只股票""" + stocks = [ + ("600519", "贵州茅台", "2023-01-01"), + ("000858", "五粮液", "2023-01-01"), + ("000001", "平安银行", "2023-01-01"), + ("300750", "宁德时代", "2023-01-01"), + ("510300", "沪深300ETF", "2023-01-01"), + ] + end = datetime.now().strftime("%Y-%m-%d") + results = [] + + print(f"\n{'='*70}") + print(f"{'股票':<12} {'基准':>8} {'MA20':>8} {'MACD':>8} {'自适应α':>10} {'自适应回撤':>10} {'自适应胜率':>8}") + print(f"{'-'*70}") + + for code, name, start in stocks: + df = get_data(code, start, end) + if df is None: + print(f"{name:<12} 数据获取失败") + continue + r = adaptive_backtest(df) + b = run_benchmark(df) + results.append({ + "name": name, "code": code, + "buyhold": b["buyhold"], + "ma20": b["ma20"], + "macd": b["macd"], + "adaptive": r["return"], + "adaptive_alpha": r["alpha"], + "adaptive_dd": r["max_dd"], + "adaptive_wr": r["winrate"], + "ma20_trades": r["ma20_trades"], + "macd_trades": r["macd_trades"], + }) + print(f"{name:<12} {b['buyhold']:>+7.1f}% {b['ma20']:>+7.1f}% {b['macd']:>+7.1f}% {r['alpha']:>+9.1f}% {r['max_dd']:>9.1f}% {r['winrate']:>7.0f}%") + + print(f"{'='*70}") + + # 汇总 + avg_alpha = np.mean([x["adaptive_alpha"] for x in results]) + win_count = sum(1 for x in results if x["adaptive_alpha"] > 0) + print(f"\n自适应策略平均α: {avg_alpha:+.1f}% | 跑赢基准: {win_count}/{len(results)}只") + + msg = f"""📊 自适应策略批量测试(5只) + +| 股票 | 基准 | MA20 | MACD | 自适应α | 回撤 | 胜率 | +|------|------|------|------|--------|------|------| +""" + for x in results: + msg += f"| {x['name']} | {x['buyhold']:+.0f}% | {x['ma20']:+.0f}% | {x['macd']:+.0f}% | **{x['adaptive_alpha']:+.1f}%** | {x['adaptive_dd']:.0f}% | {x['adaptive_wr']:.0f}% |\n" + + msg += f""" +平均超额收益(α): **{avg_alpha:+.1f}%** +跑赢基准: {win_count}/{len(results)}只 + +自适应策略: ADX>25用MA20突破, ADX≤25用MACD +结论: {'有效 ✅' if avg_alpha > 0 else '无效 ❌'} + +生成: {datetime.now().strftime('%Y-%m-%d %H:%M')}""" + send_feishu(msg) + print("\n✅ 批量测试完成，已推送飞书") + + +if __name__ == "__main__": + if "--batch" in sys.argv: + batch_test() + elif len(sys.argv) < 2: + print("用法: python3 stock_adaptive.py <代码> [起始] [结束]") + print(" python3 stock_adaptive.py --batch") + else: + code = sys.argv[1] + start = sys.argv[2] if len(sys.argv) > 2 else "2023-01-01" + end = sys.argv[3] if len(sys.argv) > 3 else datetime.now().strftime("%Y-%m-%d") + analyze(code, start, end) \ No newline at end of file