270 lines
8.8 KiB
Python
270 lines
8.8 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"
|
||
FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/446db983-e392-4d2c-bfb8-f9060e5df3ad"
|
||
|
||
|
||
def load_account():
|
||
if ACCOUNT_FILE.exists():
|
||
with open(ACCOUNT_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": "五粮液(000858)",
|
||
"start_date": "2026-07-12",
|
||
"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):
|
||
with open(ACCOUNT_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):
|
||
"""模拟买入"""
|
||
acct = load_account()
|
||
price = float(price)
|
||
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)
|
||
|
||
msg = f"""🟢 模拟买入 — 五粮液(000858)
|
||
|
||
日期: {today}
|
||
价格: {price:.2f}
|
||
数量: {shares}股
|
||
金额: {cost:.2f}
|
||
剩余现金: {acct['current_capital']:.2f}
|
||
|
||
MA20金叉触发，已记录纸上持仓。
|
||
当前持仓: {shares}股 (成本{price:.2f})
|
||
|
||
小唯股票投研 · 模拟账户"""
|
||
send_feishu(msg)
|
||
print(msg)
|
||
|
||
|
||
def cmd_sell(price):
|
||
"""模拟卖出"""
|
||
acct = load_account()
|
||
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)
|
||
|
||
msg = f"""🔴 模拟卖出 — 五粮液(000858)
|
||
|
||
日期: {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():
|
||
"""查看当前账户状态"""
|
||
acct = load_account()
|
||
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("000858")
|
||
|
||
print(f"\n{'='*50}")
|
||
print(f"小唯模拟账户 — 五粮液(000858) 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():
|
||
"""生成每日账户报告"""
|
||
acct = load_account()
|
||
stats = acct["stats"]
|
||
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("000858")
|
||
|
||
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__":
|
||
if len(sys.argv) < 2:
|
||
cmd_status()
|
||
elif sys.argv[1] == "status":
|
||
cmd_status()
|
||
elif sys.argv[1] == "buy" and len(sys.argv) >= 3:
|
||
cmd_buy(sys.argv[2])
|
||
elif sys.argv[1] == "sell" and len(sys.argv) >= 3:
|
||
cmd_sell(sys.argv[2])
|
||
elif sys.argv[1] == "report":
|
||
cmd_report()
|
||
else:
|
||
print("用法:")
|
||
print(" python3 stock_paper.py status # 账户状态")
|
||
print(" python3 stock_paper.py buy <价格> # 模拟买入")
|
||
print(" python3 stock_paper.py sell <价格> # 模拟卖出")
|
||
print(" python3 stock_paper.py report # 每日报告") |