166 lines
6.0 KiB
Python
166 lines
6.0 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
小唯股票实盘接口抽象层 — Broker 后端
|
||
==================================
|
||
统一交易接口，按配置选择后端：
|
||
|
||
1. paper 模拟盘（默认，走 stock_paper.py）
|
||
2. feishu 飞书提醒人工下单（最安全，A股散户首选）
|
||
3. easytrader 券商客户端自动下单（预留，需 Windows + 券商客户端）
|
||
|
||
用法:
|
||
python3 stock_broker.py status # 查看当前后端配置
|
||
python3 stock_broker.py execute <code> <signal> # 按信号执行一笔交易
|
||
python3 stock_broker.py test # 测试当前后端
|
||
|
||
配置文件: ~/.hermes/stock_broker.json
|
||
{"backend": "feishu"} # 默认飞书提醒（安全）
|
||
{"backend": "paper"} # 模拟盘自动成交
|
||
{"backend": "easytrader", "broker": "ht", "config": {...}} # 预留
|
||
"""
|
||
|
||
import json, sys, urllib.request
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
CONFIG_FILE = Path.home() / ".hermes" / "stock_broker.json"
|
||
FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/446db983-e392-4d2c-bfb8-f9060e5df3ad"
|
||
|
||
# 股票名称映射
|
||
STOCK_NAMES = {
|
||
"000858": "五粮液", "600519": "贵州茅台", "000568": "泸州老窖",
|
||
"002304": "洋河股份", "600036": "招商银行", "601318": "中国平安",
|
||
"000001": "平安银行",
|
||
}
|
||
|
||
|
||
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
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def load_config():
|
||
default = {"backend": "feishu"}
|
||
if CONFIG_FILE.exists():
|
||
try:
|
||
d = json.load(open(CONFIG_FILE))
|
||
d.setdefault("backend", "feishu")
|
||
return d
|
||
except Exception:
|
||
pass
|
||
return default
|
||
|
||
|
||
def save_config(cfg):
|
||
with open(CONFIG_FILE, "w") as f:
|
||
json.dump(cfg, f, ensure_ascii=False, indent=2)
|
||
|
||
|
||
# ==================== 后端实现 ====================
|
||
|
||
def backend_paper(code, signal, price):
|
||
"""模拟盘：走 stock_paper.execute_signal"""
|
||
import stock_paper
|
||
sig = {"code": code, "signal": signal, "close": price,
|
||
"name": STOCK_NAMES.get(code, code)}
|
||
action, detail = stock_paper.execute_signal(sig)
|
||
return f"[模拟盘] {action}: {detail}"
|
||
|
||
|
||
def backend_feishu(code, signal, price):
|
||
"""飞书提醒人工下单（A股散户最安全路径）"""
|
||
name = STOCK_NAMES.get(code, code)
|
||
action_text = {
|
||
"BUY": f"🟢 建议买入 {name}({code}) @ {price:.2f}",
|
||
"SELL": f"🔴 建议卖出 {name}({code}) @ {price:.2f}",
|
||
}.get(signal, f"⏸️ {name}({code}) 维持现状 ({signal})")
|
||
msg = (f"⚠️ 人工下单提醒\n{action_text}\n\n"
|
||
f"策略: MA20突破 | 触发: {datetime.now().strftime('%m-%d %H:%M')}\n"
|
||
f"请登录券商 APP 手动操作（T+1，注意仓位）")
|
||
ok = send_feishu(msg)
|
||
return f"[飞书提醒] {'已推送' if ok else '推送失败'}: {action_text}"
|
||
|
||
|
||
def backend_easytrader(code, signal, price, broker_cfg=None):
|
||
"""easytrader 券商自动下单（预留，未接入）"""
|
||
return ("[easytrader] 未接入：需要 Windows + 券商客户端 + easytrader 库。\n"
|
||
"接入步骤: 1) pip install easytrader 2) 配置 use() 券商 3) 本文件接通 trade()")
|
||
|
||
|
||
BACKENDS = {
|
||
"paper": backend_paper,
|
||
"feishu": backend_feishu,
|
||
"easytrader": backend_easytrader,
|
||
}
|
||
|
||
|
||
def execute(code, signal, price):
|
||
"""按配置后端执行一笔交易"""
|
||
cfg = load_config()
|
||
backend = cfg.get("backend", "feishu")
|
||
if backend not in BACKENDS:
|
||
return f"❌ 未知后端: {backend}（可用: paper/feishu/easytrader）"
|
||
fn = BACKENDS[backend]
|
||
if backend == "easytrader":
|
||
return fn(code, signal, price, cfg.get("config"))
|
||
return fn(code, signal, price)
|
||
|
||
|
||
def cmd_status():
|
||
cfg = load_config()
|
||
backend = cfg.get("backend", "feishu")
|
||
print(f"\n{'='*50}")
|
||
print("小唯股票实盘接口 — Broker 配置")
|
||
print(f"{'='*50}")
|
||
print(f"当前后端: {backend}")
|
||
print(f"配置文件: {CONFIG_FILE}")
|
||
print(f"\n可用后端:")
|
||
for name, desc in [("paper", "模拟盘（自动成交，练手）"),
|
||
("feishu", "飞书提醒人工下单（默认，安全）"),
|
||
("easytrader", "券商自动下单（预留，需 Windows）")]:
|
||
mark = " ✅" if name == backend else ""
|
||
print(f" {name:<12} {desc}{mark}")
|
||
print(f"\n切换: 编辑 {CONFIG_FILE} 修改 backend 字段")
|
||
print(f" echo '{{\"backend\": \"paper\"}}' > {CONFIG_FILE}")
|
||
print(f"{'='*50}")
|
||
|
||
|
||
def cmd_test():
|
||
"""测试当前后端（不发真实交易）"""
|
||
cfg = load_config()
|
||
backend = cfg.get("backend", "feishu")
|
||
print(f"测试后端: {backend}")
|
||
if backend == "paper":
|
||
import stock_paper
|
||
r = stock_paper.execute_signal({"signal": "HOLD_SHORT", "close": 78.0}, dry_run=True)
|
||
print(f" ✅ 模拟盘连通: {r}")
|
||
elif backend == "feishu":
|
||
ok = send_feishu(f"🧪 Broker 测试 {datetime.now().strftime('%H:%M')}：飞书通道正常")
|
||
print(f" {'✅ 飞书推送成功' if ok else '❌ 飞书推送失败'}")
|
||
elif backend == "easytrader":
|
||
print(" ⚠️ easytrader 未接入，无法测试")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) < 2:
|
||
cmd_status()
|
||
elif sys.argv[1] == "status":
|
||
cmd_status()
|
||
elif sys.argv[1] == "test":
|
||
cmd_test()
|
||
elif sys.argv[1] == "execute" and len(sys.argv) >= 5:
|
||
code, signal, price = sys.argv[2], sys.argv[3], float(sys.argv[4])
|
||
print(execute(code, signal, price))
|
||
else:
|
||
print("用法:")
|
||
print(" python3 stock_broker.py status # 查看配置")
|
||
print(" python3 stock_broker.py test # 测试后端")
|
||
print(" python3 stock_broker.py execute <code> <signal> <price> # 执行交易")
|