265 lines
8.8 KiB
Python
265 lines
8.8 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
小唯股票回测引擎 v3 — MACD策略
|
||
==============================
|
||
真实数据源: 腾讯/ifzq K线API (akshare备用)
|
||
策略: MACD金叉买/死叉卖
|
||
|
||
用法:
|
||
python3 stock_macd_strategy.py <股票代码> [起始日期] [结束日期]
|
||
python3 stock_macd_strategy.py --demo # 模拟数据
|
||
python3 stock_macd_strategy.py 600519 2023-01-01 2026-07-11
|
||
"""
|
||
|
||
import json, re, subprocess, sys, urllib.request
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
import pandas as pd
|
||
import numpy as np
|
||
|
||
HOME = Path.home()
|
||
OUTPUT = 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 as e:
|
||
print(f"飞书推送失败: {e}")
|
||
|
||
|
||
# ===================== 真实数据获取 =====================
|
||
|
||
def get_real_data(stock_code, start_date, end_date):
|
||
"""
|
||
腾讯/ifzq K线API，格式: sh600519 -> 返回前复权日K
|
||
返回 DataFrame 或 None(失败时)
|
||
"""
|
||
# 转换代码
|
||
if stock_code.startswith("6"):
|
||
mc = f"sh{stock_code}"
|
||
else:
|
||
mc = f"sz{stock_code}"
|
||
|
||
url = (f"https://web.ifzq.gtimg.cn/appstock/app/fqkline/get"
|
||
f"?_var=kline_dayqfq¶m={mc},day,{start_date},{end_date},500,qfq")
|
||
|
||
try:
|
||
text = urllib.request.urlopen(url, timeout=10).read().decode("utf-8")
|
||
text = text.replace("kline_dayqfq=", "", 1)
|
||
data = json.loads(text)
|
||
qfq = data.get("data", {}).get(mc, {}).get("qfqday") or data.get("data", {}).get(mc, {}).get("day") or []
|
||
if not qfq:
|
||
return None
|
||
|
||
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 as e:
|
||
print(f" ⚠️ 真实数据获取失败: {e}")
|
||
return None
|
||
|
||
|
||
# ===================== 模拟数据 =====================
|
||
|
||
def generate_synthetic(stock_code, start_date, end_date):
|
||
"""生成模拟数据(当真实数据不可用时)"""
|
||
dates = pd.date_range(start_date, end_date, freq="B")
|
||
n = len(dates)
|
||
closes = 10 * np.exp(np.cumsum(np.random.normal(0.0003, 0.02, n)))
|
||
for i in range(0, n, 20):
|
||
burst = np.random.choice([-1, 1]) * np.random.uniform(0.005, 0.015)
|
||
for j in range(min(10, n - i)):
|
||
closes[i + j] *= (1 + burst)
|
||
df = pd.DataFrame({"close": closes}, index=dates)
|
||
df["open"] = df["close"] * (1 + np.random.uniform(-0.005, 0.005, n))
|
||
df["high"] = df["close"] * (1 + np.abs(np.random.normal(0, 0.01, n)))
|
||
df["low"] = df["close"] * (1 - np.abs(np.random.normal(0, 0.01, n)))
|
||
df["volume"] = np.random.uniform(1e6, 5e6, n)
|
||
return df
|
||
|
||
|
||
# ===================== MACD 计算 =====================
|
||
|
||
def compute_macd(closes, fast=12, slow=26, signal=9):
|
||
ema_fast = pd.Series(closes).ewm(span=fast).mean()
|
||
ema_slow = pd.Series(closes).ewm(span=slow).mean()
|
||
macd = ema_fast - ema_slow
|
||
signal_line = macd.ewm(span=signal).mean()
|
||
hist = macd - signal_line
|
||
return macd, signal_line, hist
|
||
|
||
|
||
# ===================== 回测引擎 =====================
|
||
|
||
def run_backtest(df, initial_cash=100000, trade_log=None):
|
||
"""纯Python MACD回测"""
|
||
fast, slow, sig = 12, 26, 9
|
||
macd, signal_line, _ = compute_macd(df["close"])
|
||
|
||
cash = initial_cash
|
||
shares = 0
|
||
peak = initial_cash
|
||
max_dd = 0.0
|
||
trades = []
|
||
total_trades = 0
|
||
wins = 0
|
||
|
||
close_arr = df["close"].values
|
||
|
||
for i in range(slow, len(close_arr)):
|
||
price = close_arr[i]
|
||
|
||
# 金叉
|
||
if (macd.iloc[i] > signal_line.iloc[i] and
|
||
macd.iloc[i-1] <= signal_line.iloc[i-1]):
|
||
if shares == 0:
|
||
s = int(cash / price)
|
||
if s > 0:
|
||
shares = s
|
||
cash -= s * price
|
||
trades.append(("BUY", df.index[i], s, price, cash))
|
||
|
||
# 死叉
|
||
elif (macd.iloc[i] < signal_line.iloc[i] and
|
||
macd.iloc[i-1] >= signal_line.iloc[i-1]):
|
||
if shares > 0:
|
||
proceeds = shares * price
|
||
won = proceeds > (initial_cash if not trades else 0)
|
||
if won:
|
||
wins += 1
|
||
trades.append(("SELL", df.index[i], shares, price, cash + proceeds))
|
||
cash += proceeds
|
||
total_trades += 1
|
||
shares = 0
|
||
|
||
equity = cash + shares * price
|
||
peak = max(peak, equity)
|
||
dd = (peak - equity) / peak * 100 if peak > 0 else 0
|
||
max_dd = max(max_dd, dd)
|
||
|
||
final = cash + shares * close_arr[-1]
|
||
strat_ret = (final - initial_cash) / initial_cash * 100
|
||
bh_ret = (close_arr[-1] - close_arr[0]) / close_arr[0] * 100
|
||
|
||
return {
|
||
"initial_cash": initial_cash,
|
||
"final_value": final,
|
||
"strategy_return": strat_ret,
|
||
"buyhold_return": bh_ret,
|
||
"alpha": strat_ret - bh_ret,
|
||
"max_drawdown": max_dd,
|
||
"total_trades": total_trades,
|
||
"win_rate": wins / total_trades * 100 if total_trades > 0 else 0,
|
||
"trades": trades,
|
||
}
|
||
|
||
|
||
# ===================== 主流程 =====================
|
||
|
||
def analyze(stock_code, start, end=None):
|
||
end = end or datetime.now().strftime("%Y-%m-%d")
|
||
|
||
print(f"\n{'='*50}")
|
||
print(f"小唯股票回测 — MACD策略")
|
||
print(f"代码: {stock_code} | 时间: {start} ~ {end}")
|
||
print(f"{'='*50}")
|
||
|
||
# 获取数据
|
||
df = get_real_data(stock_code, start, end)
|
||
data_src = "真实" if df is not None else "模拟"
|
||
|
||
if df is None or len(df) < 60:
|
||
print(f" 📊 使用模拟数据")
|
||
df = generate_synthetic(stock_code, start, end)
|
||
else:
|
||
print(f" ✅ 使用{data_src}数据，共 {len(df)} 条")
|
||
|
||
print(f" 运行MACD回测...")
|
||
result = run_backtest(df)
|
||
|
||
# 输出
|
||
print(f"\n{'='*50} 回测结果")
|
||
print(f" 初始资金: {result['initial_cash']:.0f}")
|
||
print(f" 最终资产: {result['final_value']:.0f}")
|
||
print(f" 策略收益: {result['strategy_return']:+.2f}%")
|
||
print(f" 买入持有: {result['buyhold_return']:+.2f}%")
|
||
print(f" 超额收益(α):{result['alpha']:+.2f}%")
|
||
print(f" 最大回撤: {result['max_drawdown']:.2f}%")
|
||
print(f" 交易次数: {result['total_trades']}")
|
||
print(f" 胜率: {result['win_rate']:.1f}%")
|
||
|
||
if result["trades"]:
|
||
print(f"\n 最近5笔交易:")
|
||
for action, date, qty, price, cash_bal in result["trades"][-5:]:
|
||
print(f" [{date.date()}] {action} {qty}股@{price:.2f} 余额{cash_bal:.0f}")
|
||
|
||
# 保存
|
||
result_file = OUTPUT / f"result_{stock_code}.json"
|
||
with open(result_file, "w") as f:
|
||
json.dump({k: str(v) if k == "trades" else v for k, v in result.items()}, f, ensure_ascii=False, indent=2, default=str)
|
||
|
||
# 飞书
|
||
msg = f"""📈 小唯股票回测报告
|
||
|
||
代码: {stock_code}
|
||
策略: MACD金叉/死叉
|
||
时间: {start} ~ {end}
|
||
数据: {data_src} | {len(df)}条
|
||
|
||
初始资金: {result['initial_cash']:.0f}
|
||
最终资产: {result['final_value']:.0f}
|
||
策略收益: {result['strategy_return']:+.1f}%
|
||
买入持有: {result['buyhold_return']:+.1f}%
|
||
超额收益: {result['alpha']:+.1f}%
|
||
最大回撤: {result['max_drawdown']:.1f}%
|
||
交易次数: {result['total_trades']}
|
||
胜率: {result['win_rate']:.0f}%
|
||
|
||
生成: {datetime.now().strftime('%Y-%m-%d %H:%M')}
|
||
小唯股票投研 · 模拟阶段"""
|
||
send_feishu(msg)
|
||
print(f"\n✅ 报告已推送，数据存 {result_file}")
|
||
return result
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if "--demo" in sys.argv:
|
||
analyze("DEMO", "2023-01-01", "2026-07-11")
|
||
elif len(sys.argv) < 2:
|
||
print("用法:")
|
||
print(" python3 stock_macd_strategy.py <代码> [起始] [结束]")
|
||
print(" python3 stock_macd_strategy.py --demo")
|
||
else:
|
||
code = sys.argv[1]
|
||
s = sys.argv[2] if len(sys.argv) > 2 else "2023-01-01"
|
||
e = sys.argv[3] if len(sys.argv) > 3 else datetime.now().strftime("%Y-%m-%d")
|
||
analyze(code, s, e) |