441 lines
16 KiB
Python
441 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
小唯模拟交易账户 — 纸上盈亏追踪
|
||
=================================
|
||
MA20金叉/死叉自动记录，跟踪模拟账户表现
|
||
|
||
功能:
|
||
- 买入/卖出时自动记录
|
||
- 计算纸上盈亏/胜率/最大回撤
|
||
- 每日估值
|
||
- 历史成交记录
|
||
|
||
用法:
|
||
python3 stock_paper.py status # 查看当前账户状态
|
||
python3 stock_paper.py buy <价格> # 模拟买入（MA20金叉触发）
|
||
python3 stock_paper.py sell <价格> # 模拟卖出（MA20死叉触发）
|
||
python3 stock_paper.py report # 生成模拟账户报告
|
||
"""
|
||
|
||
import json, sys, urllib.request
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
OUTPUT = Path.home() / ".hermes" / "stock_backtest"
|
||
OUTPUT.mkdir(exist_ok=True)
|
||
ACCOUNT_FILE = OUTPUT / "paper_trades_000858.json" # 默认五粮液，execute_signal 会按代码动态覆盖
|
||
|
||
# 股票名称映射（与 stock_signal.py / stock_portfolio.py 一致）
|
||
STOCK_NAMES = {
|
||
"000858": "五粮液",
|
||
"600519": "贵州茅台",
|
||
"000568": "泸州老窖",
|
||
"002304": "洋河股份",
|
||
"600036": "招商银行",
|
||
"601318": "中国平安",
|
||
"000001": "平安银行",
|
||
"300750": "宁德时代",
|
||
"002594": "比亚迪",
|
||
"002415": "海康威视",
|
||
"600030": "中信证券",
|
||
"601899": "紫金矿业",
|
||
"601857": "中国石油",
|
||
"601088": "中国神华",
|
||
"600188": "兖矿能源",
|
||
"688981": "中芯国际",
|
||
"000333": "美的集团",
|
||
"600276": "恒瑞医药",
|
||
"600941": "中国移动",
|
||
"000002": "万科A",
|
||
"601012": "隆基绿能",
|
||
"000063": "中兴通讯",
|
||
"600900": "长江电力",
|
||
"601398": "工商银行",
|
||
"601166": "兴业银行",
|
||
}
|
||
|
||
def _account_file(code=None):
|
||
"""根据股票代码返回账户文件路径，None 时用默认五粮液"""
|
||
if code:
|
||
return OUTPUT / f"paper_trades_{code}.json"
|
||
return ACCOUNT_FILE
|
||
FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/446db983-e392-4d2c-bfb8-f9060e5df3ad"
|
||
|
||
|
||
def load_account(code=None):
|
||
acct_file = _account_file(code)
|
||
if acct_file.exists():
|
||
with open(acct_file) as f:
|
||
d = json.load(f)
|
||
# 确保字段完整(兼容旧格式)
|
||
d.setdefault("current_capital", d.get("current_capital", 100000))
|
||
d.setdefault("positions", d.get("positions", []))
|
||
d.setdefault("closed_trades", d.get("closed_trades", []))
|
||
d.setdefault("last_signal", "空仓")
|
||
d.setdefault("last_signal_date", "")
|
||
d.setdefault("stats", d.get("stats", {"total_trades": 0, "winning_trades": 0, "losing_trades": 0}))
|
||
return d
|
||
return {
|
||
"strategy": "MA20突破",
|
||
"stock": STOCK_NAMES.get(code, f"未知({code})") if code else "五粮液(000858)",
|
||
"stock_code": code or "000858",
|
||
"start_date": datetime.now().strftime("%Y-%m-%d"),
|
||
"initial_capital": 100000,
|
||
"current_capital": 100000,
|
||
"positions": [], # [{"shares": N, "avg_cost": P}]
|
||
"closed_trades": [], # [{"date": "", "buy_price": P, "sell_price": P, "shares": N, "pnl": P}]
|
||
"last_signal": "空仓",
|
||
"last_signal_date": "",
|
||
"stats": {"total_trades": 0, "winning_trades": 0, "losing_trades": 0}
|
||
}
|
||
|
||
|
||
def save_account(acct, code=None):
|
||
acct_file = _account_file(code)
|
||
with open(acct_file, "w") as f:
|
||
json.dump(acct, f, ensure_ascii=False, indent=2)
|
||
|
||
|
||
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_current_price(code="000858"):
|
||
mc = f"sz{code}" if not code.startswith("sh") else code
|
||
try:
|
||
url = f"https://qt.gtimg.cn/q={mc}"
|
||
text = urllib.request.urlopen(url, timeout=3).read().decode("gbk")
|
||
parts = text.split("~")
|
||
if len(parts) > 10:
|
||
return float(parts[3]), float(parts[4]) # 当前价, 昨收
|
||
except Exception:
|
||
pass
|
||
return None, None
|
||
|
||
|
||
def cmd_buy(price, shares=None, code=None):
|
||
"""模拟买入"""
|
||
acct = load_account(code)
|
||
price = float(price)
|
||
name = STOCK_NAMES.get(code, acct.get("stock", "五粮液")) if code else acct.get("stock", "五粮液")
|
||
shares = shares or int(acct["current_capital"] // price)
|
||
cost = shares * price
|
||
|
||
if cost > acct["current_capital"]:
|
||
shares = int(acct["current_capital"] // price)
|
||
cost = shares * price
|
||
|
||
if shares <= 0:
|
||
print("资金不足，无法买入")
|
||
return
|
||
|
||
today = datetime.now().strftime("%Y-%m-%d")
|
||
acct["positions"].append({"shares": shares, "avg_cost": price})
|
||
acct["current_capital"] -= cost
|
||
acct["last_signal"] = "买入"
|
||
acct["last_signal_date"] = today
|
||
save_account(acct, code)
|
||
|
||
msg = f"""🟢 模拟买入 — {name}
|
||
|
||
日期: {today}
|
||
价格: {price:.2f}
|
||
数量: {shares}股
|
||
金额: {cost:.2f}
|
||
剩余现金: {acct['current_capital']:.2f}
|
||
|
||
MA20金叉触发，已记录纸上持仓。
|
||
当前持仓: {shares}股 (成本{price:.2f})
|
||
|
||
小唯股票投研 · 模拟账户"""
|
||
send_feishu(msg)
|
||
print(msg)
|
||
|
||
|
||
def execute_signal(sig, dry_run=False):
|
||
"""
|
||
信号自动执行：把技术信号真正落地为模拟交易（防重复）
|
||
|
||
规则:
|
||
- BUY: 无持仓才买（避免重复买入）
|
||
- SELL: 有持仓才卖（避免空卖）
|
||
- HOLD_LONG / HOLD_SHORT: 不动
|
||
- 弱势行业 BUY 半仓买入（2026-08-01 回测优化：全拦截损失 α，半仓最优）
|
||
|
||
返回: (action, detail)
|
||
"""
|
||
if not sig:
|
||
return ("SKIP", "无信号")
|
||
|
||
# 关键修复：从信号中提取股票代码，动态定位正确的账户文件
|
||
code = sig.get("code") or sig.get("stock_code", "000858")
|
||
acct = load_account(code)
|
||
has_position = bool(acct["positions"])
|
||
signal = sig.get("signal")
|
||
|
||
# v2：弱势行业金叉半仓买入（2026-08-01 回测优化：全拦截损失 α，半仓最优）
|
||
weak_industries = {"白酒", "医药", "通信", "汽车", "地产"}
|
||
is_weak = signal == "BUY" and sig.get("industry") in weak_industries
|
||
|
||
if signal == "BUY" and not has_position:
|
||
price = sig.get("close") or sig.get("price")
|
||
name = STOCK_NAMES.get(code, code)
|
||
if dry_run:
|
||
mode = "半仓(弱势行业)" if is_weak else "全仓"
|
||
print(f"[DRY-RUN] {name} 金叉{mode}买入 @ {price:.2f}")
|
||
return ("BUY", f"{name} {mode}金叉买入@{price:.2f}")
|
||
if is_weak:
|
||
# 半仓：只用一半资金
|
||
shares = int((acct["current_capital"] * 0.5) // price)
|
||
if shares <= 0:
|
||
return ("SKIP", "资金不足")
|
||
cost = shares * price
|
||
acct["positions"].append({"shares": shares, "avg_cost": price})
|
||
acct["current_capital"] -= cost
|
||
acct["last_signal"] = "半仓买入"
|
||
acct["last_signal_date"] = datetime.now().strftime("%Y-%m-%d")
|
||
save_account(acct, code)
|
||
msg = (f"⚠️ 模拟半仓买入 — {name} 弱势行业金叉\n\n"
|
||
f"日期: {datetime.now().strftime('%Y-%m-%d')}\n"
|
||
f"价格: {price:.2f}\n数量: {shares}股\n金额: {cost:.2f}\n"
|
||
f"剩余现金: {acct['current_capital']:.2f}\n\n"
|
||
f"弱势行业金叉，按半仓规则买入。")
|
||
send_feishu(msg)
|
||
print(msg)
|
||
return ("BUY_HALF", f"{name} 弱势行业半仓买入@{price:.2f}")
|
||
# 全仓买入
|
||
shares = int(acct["current_capital"] // price)
|
||
if shares <= 0:
|
||
return ("SKIP", "资金不足")
|
||
cost = shares * price
|
||
acct["positions"].append({"shares": shares, "avg_cost": price})
|
||
acct["current_capital"] -= cost
|
||
acct["last_signal"] = "买入"
|
||
acct["last_signal_date"] = datetime.now().strftime("%Y-%m-%d")
|
||
save_account(acct, code)
|
||
msg = (f"🟢 模拟买入 — {name}({code})\n\n"
|
||
f"日期: {datetime.now().strftime('%Y-%m-%d')}\n"
|
||
f"价格: {price:.2f}\n数量: {shares}股\n金额: {cost:.2f}\n"
|
||
f"剩余现金: {acct['current_capital']:.2f}\n\n"
|
||
f"MA20金叉触发，已记录纸上持仓。")
|
||
send_feishu(msg)
|
||
print(msg)
|
||
return ("BUY", f"{name} 金叉买入@{price:.2f}")
|
||
|
||
elif signal == "SELL" and has_position:
|
||
price = sig.get("close") or sig.get("price")
|
||
name = STOCK_NAMES.get(code, code)
|
||
if dry_run:
|
||
print(f"[DRY-RUN] {name} 死叉卖出 @ {price:.2f}")
|
||
return ("SELL", f"{name} 死叉卖出@{price:.2f}")
|
||
# 卖出：平掉所有持仓
|
||
total_shares = sum(p["shares"] for p in acct["positions"])
|
||
total_cost = sum(p["shares"] * p["avg_cost"] for p in acct["positions"])
|
||
revenue = total_shares * price
|
||
pnl = revenue - total_cost
|
||
acct["current_capital"] += revenue
|
||
acct["positions"] = []
|
||
acct["last_signal"] = "卖出"
|
||
acct["last_signal_date"] = datetime.now().strftime("%Y-%m-%d")
|
||
acct["closed_trades"].append({
|
||
"date": datetime.now().strftime("%Y-%m-%d"),
|
||
"buy_price": total_cost / total_shares if total_shares else 0,
|
||
"sell_price": price,
|
||
"shares": total_shares,
|
||
"pnl": pnl,
|
||
})
|
||
acct["stats"]["total_trades"] += 1
|
||
if pnl > 0:
|
||
acct["stats"]["winning_trades"] += 1
|
||
else:
|
||
acct["stats"]["losing_trades"] += 1
|
||
acct["stats"]["total_pnl"] = acct["stats"].get("total_pnl", 0) + pnl
|
||
save_account(acct, code)
|
||
msg = (f"🔴 模拟卖出 — {name}({code})\n\n"
|
||
f"日期: {datetime.now().strftime('%Y-%m-%d')}\n"
|
||
f"卖出价: {price:.2f}\n数量: {total_shares}股\n"
|
||
f"盈亏: {pnl:+.2f}元\n\n"
|
||
f"MA20死叉触发，已平仓。")
|
||
send_feishu(msg)
|
||
print(msg)
|
||
return ("SELL", f"{name} 死叉卖出@{price:.2f} 盈亏{pnl:+.0f}")
|
||
|
||
elif signal == "BUY" and has_position:
|
||
return ("HOLD", "已持仓，忽略重复买入")
|
||
|
||
elif signal == "SELL" and not has_position:
|
||
return ("HOLD", "已空仓，忽略重复卖出")
|
||
|
||
elif signal == "HOLD_LONG":
|
||
if has_position:
|
||
return ("HOLD", "持仓中，继续持有")
|
||
return ("HOLD", "信号建议持仓，但账户空仓（非金叉不自动买入）")
|
||
|
||
else:
|
||
return ("HOLD", "空仓观望")
|
||
|
||
|
||
def cmd_sell(price, code=None):
|
||
"""模拟卖出"""
|
||
acct = load_account(code)
|
||
name = STOCK_NAMES.get(code, acct.get("stock", "五粮液")) if code else acct.get("stock", "五粮液")
|
||
price = float(price)
|
||
today = datetime.now().strftime("%Y-%m-%d")
|
||
|
||
if not acct["positions"]:
|
||
print("没有持仓，无法卖出")
|
||
return
|
||
|
||
total_shares = sum(p["shares"] for p in acct["positions"])
|
||
total_cost = sum(p["shares"] * p["avg_cost"] for p in acct["positions"])
|
||
avg_cost = total_cost / total_shares
|
||
|
||
pnl = (price - avg_cost) * total_shares
|
||
pnl_pct = (price - avg_cost) / avg_cost * 100
|
||
|
||
acct["closed_trades"].append({
|
||
"date": today,
|
||
"buy_price": avg_cost,
|
||
"sell_price": price,
|
||
"shares": total_shares,
|
||
"pnl": pnl,
|
||
"pnl_pct": pnl_pct
|
||
})
|
||
acct["current_capital"] += total_shares * price
|
||
acct["positions"] = []
|
||
acct["last_signal"] = "卖出"
|
||
acct["last_signal_date"] = today
|
||
|
||
stats = acct["stats"]
|
||
stats["total_trades"] += 1
|
||
if pnl > 0:
|
||
stats["winning_trades"] += 1
|
||
else:
|
||
stats["losing_trades"] += 1
|
||
|
||
save_account(acct, code)
|
||
|
||
msg = f"""🔴 模拟卖出 — {name}
|
||
|
||
日期: {today}
|
||
卖出价格: {price:.2f}
|
||
数量: {total_shares}股
|
||
买入均价: {avg_cost:.2f}
|
||
盈亏: {pnl:+.2f} ({pnl_pct:+.1f}%)
|
||
|
||
MA20死叉触发，已平仓。
|
||
现金: {acct['current_capital']:.2f}
|
||
|
||
小唯股票投研 · 模拟账户"""
|
||
send_feishu(msg)
|
||
print(msg)
|
||
|
||
|
||
def cmd_status(code=None):
|
||
"""查看当前账户状态"""
|
||
acct = load_account(code)
|
||
name = acct.get("stock", "五粮液")
|
||
stock_code = code or acct.get("stock_code", "000858")
|
||
total_shares = sum(p["shares"] for p in acct["positions"])
|
||
total_cost = sum(p["shares"] * p["avg_cost"] for p in acct["positions"])
|
||
current_price, _ = get_current_price(stock_code)
|
||
|
||
print(f"\n{'='*50}")
|
||
print(f"小唯模拟账户 — {name} MA20突破策略")
|
||
print(f"{'='*50}")
|
||
print(f"初始资金: {acct['initial_capital']:.2f}")
|
||
print(f"当前现金: {acct['current_capital']:.2f}")
|
||
print(f"持仓状态: {'空仓' if total_shares == 0 else f'持仓{total_shares}股'}")
|
||
print(f"最后信号: {acct['last_signal']} ({acct['last_signal_date']})")
|
||
|
||
if total_shares > 0 and current_price:
|
||
avg_cost = total_cost / total_shares
|
||
market_val = total_shares * current_price
|
||
unreal_pnl = market_val - total_cost
|
||
unreal_pnl_pct = unreal_pnl / total_cost * 100
|
||
print(f"\n持仓详情:")
|
||
print(f" 数量: {total_shares}股")
|
||
print(f" 成本价: {avg_cost:.2f}")
|
||
print(f" 当前价: {current_price:.2f}")
|
||
print(f" 市值: {market_val:.2f}")
|
||
print(f" 纸上盈亏: {unreal_pnl:+.2f} ({unreal_pnl_pct:+.1f}%)")
|
||
|
||
# 总资产
|
||
total = acct["current_capital"] + market_val
|
||
total_pnl = total - acct["initial_capital"]
|
||
total_pnl_pct = total_pnl / acct["initial_capital"] * 100
|
||
print(f"\n总资产: {total:.2f} (初始{acct['initial_capital']:.2f})")
|
||
print(f"总盈亏: {total_pnl:+.2f} ({total_pnl_pct:+.1f}%)")
|
||
|
||
# 历史成交
|
||
stats = acct["stats"]
|
||
print(f"\n历史成交: {stats['total_trades']}笔 | 胜{stats['winning_trades']} 负{stats['losing_trades']}")
|
||
for t in acct["closed_trades"]:
|
||
print(f" {t['date']} 买{t['buy_price']:.2f}→卖{t['sell_price']:.2f} {t['pnl']:+.2f}({t['pnl_pct']:+.1f}%)")
|
||
print(f"{'='*50}")
|
||
|
||
return acct
|
||
|
||
|
||
def cmd_report(code=None):
|
||
"""生成每日账户报告"""
|
||
acct = load_account(code)
|
||
stats = acct["stats"]
|
||
name = acct.get("stock", "五粮液")
|
||
stock_code = code or acct.get("stock_code", "000858")
|
||
total_shares = sum(p["shares"] for p in acct["positions"])
|
||
total_cost = sum(p["shares"] * p["avg_cost"] for p in acct["positions"])
|
||
current_price, _ = get_current_price(stock_code)
|
||
|
||
if total_shares > 0 and current_price:
|
||
market_val = total_shares * current_price
|
||
unreal = market_val - total_cost
|
||
total = acct["current_capital"] + market_val
|
||
else:
|
||
unreal = 0
|
||
total = acct["current_capital"]
|
||
|
||
total_pnl = total - acct["initial_capital"]
|
||
total_pnl_pct = total_pnl / acct["initial_capital"] * 100
|
||
|
||
msg = f"""📊 模拟账户日报 — 五粮液(000858)
|
||
|
||
总资产: {total:.2f} | 盈亏: {total_pnl:+.2f} ({total_pnl_pct:+.1f}%)
|
||
现金: {acct['current_capital']:.2f} | {'空仓' if total_shares == 0 else f'持仓{total_shares}股'}
|
||
|
||
{'纸上浮盈: ' + f'{unreal:+.2f}' if total_shares > 0 else ''}
|
||
胜率: {stats['winning_trades']}/{stats['total_trades']}
|
||
|
||
生成: {datetime.now().strftime('%Y-%m-%d %H:%M')}"""
|
||
send_feishu(msg)
|
||
print(msg)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
code = None
|
||
if "--code" in sys.argv:
|
||
idx = sys.argv.index("--code")
|
||
if idx + 1 < len(sys.argv):
|
||
code = sys.argv[idx + 1]
|
||
|
||
if len(sys.argv) < 2 or sys.argv[1] == "status":
|
||
cmd_status(code)
|
||
elif sys.argv[1] == "buy" and len(sys.argv) >= 3:
|
||
cmd_buy(sys.argv[2], code=code)
|
||
elif sys.argv[1] == "sell" and len(sys.argv) >= 3:
|
||
cmd_sell(sys.argv[2], code=code)
|
||
elif sys.argv[1] == "report":
|
||
cmd_report(code)
|
||
else:
|
||
print("用法:")
|
||
print(" python3 stock_paper.py status [--code 000001]")
|
||
print(" python3 stock_paper.py buy <价格> [--code 000001]")
|
||
print(" python3 stock_paper.py sell <价格> [--code 000001]")
|
||
print(" python3 stock_paper.py report [--code 000001]") |