347 lines
14 KiB
Python
347 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
全球多市场模拟交易 — 港股 + 美股（做多方向）
|
||
====================================================
|
||
2026-08-02 新增（金融工具多样性 — 多市场）
|
||
|
||
核心逻辑：MA20 策略复用到港股/美股（已验证 K 线数据可用）：
|
||
港股: hk00700 腾讯 / hk03690 美团 / hk09988 阿里（港股通标的）
|
||
美股: usAAPL.OQ 苹果 / usTSLA.OQ 特斯拉 / usNVDA.OQ 英伟达
|
||
|
||
⚠️ 时差注意：美股盘 22:30-05:00（北京时间），只做日线级扫描（收盘后），
|
||
不做盘中监测。港股盘 9:30-16:00 与 A 股重叠，可盘中。
|
||
|
||
用法:
|
||
python3 stock_global.py status # 全部全球账户状态
|
||
python3 stock_global.py scan # 扫描金叉/死叉信号
|
||
python3 stock_global.py report # 汇总报告
|
||
python3 stock_global.py backtest <market> <code> <name> # 单只回测
|
||
"""
|
||
import json, os, subprocess, sys, time
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
HOME = Path.home()
|
||
OUTPUT = HOME / ".hermes" / "stock_backtest" / "global_account"
|
||
OUTPUT.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 全球市场标的（已验证 K 线数据可用）
|
||
# ⚠️ 2026-08-02 回测教训：MA20 在港股/美股整体跑输买入持有（成熟市场趋势强，
|
||
# 频繁进出磨损收益）。策略有市场适用边界，不能盲目复制 A 股经验：
|
||
# - 买入持有更优: 腾讯(-17α) 美团(-43α) 阿里(-5α) 英伟达(-5α) 苹果(-0.4α)
|
||
# - MA20 有正 α 仅: 特斯拉(+11.6α，高波动个股)
|
||
# → 港股/美股用"买入持有为主 + 高波动个股 MA20"混合策略：
|
||
# strategy = "buyhold"（默认）或 "ma20"（仅特斯拉）
|
||
GLOBAL_TARGETS = {
|
||
# 港股（腾讯接口 hk 前缀）— 买入持有
|
||
"港股腾讯": {"code": "00700", "name": "腾讯控股", "market": "hk", "capital": 100000, "strategy": "buyhold"},
|
||
"港股美团": {"code": "03690", "name": "美团", "market": "hk", "capital": 100000, "strategy": "buyhold"},
|
||
"港股阿里": {"code": "09988", "name": "阿里巴巴", "market": "hk", "capital": 100000, "strategy": "buyhold"},
|
||
# 美股（腾讯接口 usXXX.OQ 格式）— 高波动个股用 MA20
|
||
"美股苹果": {"code": "AAPL.OQ", "name": "苹果", "market": "us", "capital": 100000, "strategy": "buyhold"},
|
||
"美股特斯拉": {"code": "TSLA.OQ", "name": "特斯拉", "market": "us", "capital": 100000, "strategy": "ma20"},
|
||
"美股英伟达": {"code": "NVDA.OQ", "name": "英伟达", "market": "us", "capital": 100000, "strategy": "buyhold"},
|
||
}
|
||
|
||
MA_PERIOD = 20
|
||
|
||
|
||
def get_url(url, timeout=8, enc="utf-8", retries=1):
|
||
env = dict(os.environ)
|
||
for k in ["http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY"]:
|
||
env.pop(k, None)
|
||
for attempt in range(1 + retries):
|
||
try:
|
||
r = subprocess.run(["curl", "-s", "--max-time", str(timeout), "--compressed", url],
|
||
capture_output=True, timeout=timeout+2, env=env)
|
||
text = r.stdout.decode(enc, errors="ignore")
|
||
if text.strip():
|
||
return text
|
||
except Exception:
|
||
pass
|
||
if attempt < retries:
|
||
time.sleep(5)
|
||
return ""
|
||
|
||
|
||
def market_code(market, code):
|
||
"""市场 → 腾讯行情代码"""
|
||
if market == "hk":
|
||
return f"hk{code}"
|
||
if market == "us":
|
||
return f"us{code}"
|
||
return ("sh" if code.startswith(("6", "5")) else "sz") + code
|
||
|
||
|
||
def get_klines(market, code, days=120):
|
||
"""获取日K前复权 → [(date, close), ...]"""
|
||
mc = market_code(market, code)
|
||
today = datetime.now().strftime("%Y-%m-%d")
|
||
url = (f"https://web.ifzq.gtimg.cn/appstock/app/fqkline/get?"
|
||
f"_var=kline_dayqfq¶m={mc},day,2026-01-01,{today},{days},qfq")
|
||
raw = get_url(url)
|
||
if "=" not in raw:
|
||
return []
|
||
try:
|
||
d = json.loads(raw[raw.index("="):].lstrip("="))
|
||
data = d.get("data", {}).get(mc, {})
|
||
kl = data.get("qfqday") or data.get("day", [])
|
||
return [(row[0], float(row[2])) for row in kl]
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
def get_quote(market, code):
|
||
mc = market_code(market, code)
|
||
raw = get_url(f"https://qt.gtimg.cn/q={mc}", enc="gbk")
|
||
if "~" not in raw:
|
||
return None
|
||
parts = raw.split("~")
|
||
return {"price": float(parts[3]), "name": parts[1]}
|
||
|
||
|
||
def account_key(market, code):
|
||
return f"{market}_{code.replace('.', '_')}"
|
||
|
||
|
||
def account_file(key):
|
||
return OUTPUT / f"global_{key}.json"
|
||
|
||
|
||
def load_account(key):
|
||
f = account_file(key)
|
||
if f.exists():
|
||
return json.load(open(f))
|
||
return {
|
||
"key": key,
|
||
"stock": "",
|
||
"code": "",
|
||
"market": "",
|
||
"initial_capital": 100000,
|
||
"current_capital": 100000,
|
||
"positions": [],
|
||
"closed_trades": [],
|
||
"last_signal": "空仓",
|
||
"last_signal_date": "",
|
||
"stats": {"total_trades": 0, "winning_trades": 0, "losing_trades": 0},
|
||
"created": datetime.now().strftime("%Y-%m-%d"),
|
||
}
|
||
|
||
|
||
def save_account(acct):
|
||
with open(account_file(acct["key"]), "w") as f:
|
||
json.dump(acct, f, ensure_ascii=False, indent=2)
|
||
|
||
|
||
def compute_ma20_signal(klines):
|
||
"""金叉/死叉信号：金叉买入（做多）"""
|
||
if len(klines) < MA_PERIOD + 2:
|
||
return "HOLD", 0, 0
|
||
closes = [k[1] for k in klines]
|
||
prev_close = closes[-2]
|
||
cur_close = closes[-1]
|
||
prev_ma20 = sum(closes[-(MA_PERIOD+1):-1]) / MA_PERIOD
|
||
cur_ma20 = sum(closes[-MA_PERIOD:]) / MA_PERIOD
|
||
prev_above = prev_close > prev_ma20
|
||
cur_above = cur_close > cur_ma20
|
||
if not prev_above and cur_above:
|
||
return "BUY", cur_close, cur_ma20
|
||
if prev_above and not cur_above:
|
||
return "SELL", cur_close, cur_ma20
|
||
return "HOLD", cur_close, cur_ma20
|
||
|
||
|
||
def execute_signal(sig, dry_run=False):
|
||
"""信号执行: sig = {"market": "hk", "code": "00700", "signal": "BUY"/"SELL", "close": price}
|
||
strategy 支持：
|
||
- buyhold: 买入后长期持有（死叉不卖，除非信号标 HOLD_BUYHOLD 强制）
|
||
- ma20: 金叉买/死叉卖（仅特斯拉）
|
||
"""
|
||
market = sig.get("market")
|
||
code = sig.get("code")
|
||
key = account_key(market, code)
|
||
acct = load_account(key)
|
||
cfg = next((c for c in GLOBAL_TARGETS.values()
|
||
if c["market"] == market and c["code"] == code), {})
|
||
strategy = cfg.get("strategy", "buyhold")
|
||
if not acct["stock"]:
|
||
acct["stock"] = cfg.get("name", "")
|
||
acct["code"] = code
|
||
acct["market"] = market
|
||
acct["strategy"] = strategy
|
||
signal = sig.get("signal")
|
||
price = sig.get("close", 0)
|
||
|
||
# 买入持有策略：死叉信号 → 忽略（长期持有，除非持仓已卖出）
|
||
if strategy == "buyhold" and signal == "SELL" and acct["positions"]:
|
||
return "HOLD", f"买入持有策略：死叉忽略，继续持有 {acct['stock']}"
|
||
|
||
if signal == "BUY":
|
||
if acct["positions"]:
|
||
return "SKIP", "已有持仓"
|
||
shares = int(acct["current_capital"] // price)
|
||
if shares <= 0:
|
||
return "SKIP", "资金不足"
|
||
if dry_run:
|
||
return "BUY", f"[DRY-RUN] 买入 {shares}股 @{price:.2f}"
|
||
acct["positions"].append({"shares": shares, "avg_cost": price,
|
||
"open_date": datetime.now().strftime("%Y-%m-%d")})
|
||
acct["current_capital"] -= shares * price # 2026-08-28 修复：买入扣减现金（之前不扣导致资产虚高）
|
||
acct["last_signal"] = f"买入 {shares}股 @{price:.2f}"
|
||
acct["last_signal_date"] = datetime.now().strftime("%Y-%m-%d")
|
||
save_account(acct)
|
||
return "BUY", f"买入 {shares}股 @{price:.2f}"
|
||
if signal == "SELL":
|
||
if not acct["positions"]:
|
||
return "SKIP", "空仓无持仓"
|
||
pos = acct["positions"][0]
|
||
shares = pos["shares"]
|
||
pnl = (price - pos["avg_cost"]) * shares
|
||
pnl_pct = pnl / (pos["avg_cost"] * shares) * 100
|
||
if dry_run:
|
||
return "SELL", f"[DRY-RUN] 卖出 {shares}股 盈亏{pnl:+,.0f} ({pnl_pct:+.2f}%)"
|
||
acct["closed_trades"].append({
|
||
"shares": shares, "avg_cost": pos["avg_cost"], "close_price": price,
|
||
"pnl": round(pnl, 2), "open_date": pos["open_date"],
|
||
"close_date": datetime.now().strftime("%Y-%m-%d"),
|
||
})
|
||
acct["current_capital"] += price * shares # 2026-08-28 修复：卖出回收全部现金（原 +pnl 在买入不扣时才是对的，买入扣款后必须 +卖出收入）
|
||
acct["positions"] = []
|
||
acct["stats"]["total_trades"] += 1
|
||
if pnl >= 0:
|
||
acct["stats"]["winning_trades"] += 1
|
||
else:
|
||
acct["stats"]["losing_trades"] += 1
|
||
acct["last_signal"] = f"卖出 盈亏{pnl:+,.0f} ({pnl_pct:+.2f}%)"
|
||
acct["last_signal_date"] = datetime.now().strftime("%Y-%m-%d")
|
||
save_account(acct)
|
||
return "SELL", f"卖出 {shares}股 盈亏{pnl:+,.0f} ({pnl_pct:+.2f}%)"
|
||
return "SKIP", f"未知信号 {signal}"
|
||
|
||
|
||
def scan_signals():
|
||
results = []
|
||
for key, cfg in GLOBAL_TARGETS.items():
|
||
klines = get_klines(cfg["market"], cfg["code"])
|
||
if not klines:
|
||
results.append({"key": key, "market": cfg["market"], "code": cfg["code"],
|
||
"name": cfg["name"], "signal": "DATA_ERR", "close": 0})
|
||
continue
|
||
sig, close, ma20 = compute_ma20_signal(klines)
|
||
dev = (close - ma20) / ma20 * 100 if ma20 else 0
|
||
# buyhold 策略：死叉不提示卖出，只提示 BUY（建仓）；持仓状态由账户决定
|
||
if cfg.get("strategy") == "buyhold" and sig == "SELL":
|
||
sig = "HOLD"
|
||
results.append({"key": key, "market": cfg["market"], "code": cfg["code"],
|
||
"name": cfg["name"], "signal": sig, "close": close,
|
||
"ma20": ma20, "dev": dev})
|
||
return results
|
||
|
||
|
||
def backtest(market, code, name):
|
||
"""MA20 回测（做多方向）"""
|
||
klines = get_klines(market, code, 120)
|
||
if len(klines) < MA_PERIOD + 2:
|
||
return None
|
||
closes = [k[1] for k in klines]
|
||
dates = [k[0] for k in klines]
|
||
capital = 100000
|
||
in_pos = False
|
||
buy_price = 0
|
||
buy_date = ""
|
||
trades = []
|
||
for i in range(MA_PERIOD, len(closes)):
|
||
prev_close = closes[i-1]
|
||
cur_close = closes[i]
|
||
prev_ma20 = sum(closes[i-MA_PERIOD:i]) / MA_PERIOD
|
||
cur_ma20 = sum(closes[i-MA_PERIOD+1:i+1]) / MA_PERIOD
|
||
prev_above = prev_close > prev_ma20
|
||
cur_above = cur_close > cur_ma20
|
||
if not prev_above and cur_above and not in_pos:
|
||
in_pos = True
|
||
buy_price = cur_close
|
||
buy_date = dates[i]
|
||
elif prev_above and not cur_above and in_pos:
|
||
pnl = (cur_close - buy_price) * int(100000 / buy_price)
|
||
trades.append({"buy": buy_date, "sell": dates[i], "buy_p": buy_price,
|
||
"sell_p": cur_close, "pnl": pnl})
|
||
capital += pnl
|
||
in_pos = False
|
||
total_pnl = capital - 100000
|
||
win = sum(1 for t in trades if t["pnl"] > 0)
|
||
# 买入持有对比
|
||
buyhold = (closes[-1] - closes[MA_PERIOD]) / closes[MA_PERIOD] * 100
|
||
return {
|
||
"code": code, "name": name, "market": market,
|
||
"strategy_return": total_pnl / 100000 * 100,
|
||
"buyhold_return": buyhold,
|
||
"alpha": total_pnl / 100000 * 100 - buyhold,
|
||
"total_trades": len(trades),
|
||
"win_rate": win / len(trades) * 100 if trades else 0,
|
||
"total_pnl": total_pnl,
|
||
"in_position": in_pos,
|
||
"last_trades": trades[-3:],
|
||
}
|
||
|
||
|
||
def cmd_status():
|
||
print(f"小唯全球市场模拟 — {datetime.now().strftime('%Y-%m-%d %H:%M')}")
|
||
print("=" * 52)
|
||
total_cap = 0
|
||
for key, cfg in GLOBAL_TARGETS.items():
|
||
acct = load_account(account_key(cfg["market"], cfg["code"]))
|
||
total_cap += acct["current_capital"]
|
||
m = {"hk": "港股", "us": "美股"}.get(cfg["market"], cfg["market"])
|
||
pos = acct["positions"]
|
||
if pos:
|
||
p = pos[0]
|
||
print(f"【{m}·{cfg['name']}】🟢持仓 {p['shares']}股 @{p['avg_cost']:.2f}")
|
||
else:
|
||
print(f"【{m}·{cfg['name']}】⚪空仓 资金{acct['current_capital']:,.0f}")
|
||
print(f"总资金: {total_cap:,.0f}")
|
||
|
||
|
||
def cmd_report():
|
||
print(f"📊 全球市场日报 {datetime.now().strftime('%Y-%m-%d')}")
|
||
total_cap = 0
|
||
total_win = total_lose = 0
|
||
for key, cfg in GLOBAL_TARGETS.items():
|
||
acct = load_account(account_key(cfg["market"], cfg["code"]))
|
||
total_cap += acct["current_capital"]
|
||
total_win += acct["stats"]["winning_trades"]
|
||
total_lose += acct["stats"]["losing_trades"]
|
||
m = {"hk": "港股", "us": "美股"}.get(cfg["market"], cfg["market"])
|
||
mark = "🟢" if acct["positions"] else "⚪"
|
||
print(f"{mark} {m}·{cfg['name']}: 资产{acct['current_capital']:,.0f} "
|
||
f"(交易{acct['stats']['total_trades']} 胜{acct['stats']['winning_trades']} 负{acct['stats']['losing_trades']})")
|
||
print(f"💰 总资产: {total_cap:,.0f}")
|
||
if total_win + total_lose > 0:
|
||
print(f"📈 胜率: {total_win/(total_win+total_lose)*100:.0f}%")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
cmd = sys.argv[1] if len(sys.argv) > 1 else "status"
|
||
if cmd == "status":
|
||
cmd_status()
|
||
elif cmd == "report":
|
||
cmd_report()
|
||
elif cmd == "scan":
|
||
for s in scan_signals():
|
||
mark = {"BUY": "🟢金叉", "SELL": "🔴死叉", "HOLD": "⚪持有", "DATA_ERR": "❌数据"}.get(s["signal"], "?")
|
||
print(f"{mark} {s['market']}·{s['name']}: 收盘{s.get('close',0):.2f} MA20{s.get('ma20',0):.2f} 偏离{s.get('dev',0):+.2f}%")
|
||
elif cmd == "backtest" and len(sys.argv) >= 4:
|
||
market = sys.argv[2]
|
||
r = backtest(market, sys.argv[3], sys.argv[4] if len(sys.argv) > 4 else sys.argv[3])
|
||
if r:
|
||
print(f"MA20回测 {r['name']}({r['code']}) [{r['market']}]")
|
||
print(f" 策略收益: {r['strategy_return']:+.2f}% | 买入持有: {r['buyhold_return']:+.2f}% | α: {r['alpha']:+.2f}%")
|
||
print(f" 交易{r['total_trades']}次 | 胜率{r['win_rate']:.0f}% | 总盈亏{r['total_pnl']:+,.0f}")
|
||
if r["last_trades"]:
|
||
print(" 最近交易:")
|
||
for t in r["last_trades"]:
|
||
print(f" {t['buy']} 买@{t['buy_p']:.2f} → {t['sell']} 卖@{t['sell_p']:.2f} 盈亏{t['pnl']:+,.0f}")
|
||
else:
|
||
print("数据不足")
|
||
else:
|
||
print(__doc__)
|