333 lines
13 KiB
Python
333 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
模拟做空系统 — 弱势行业死叉 → 做空，金叉 → 平空
|
||
====================================================
|
||
2026-08-02 新增（金融工具多样性 — 做空方向）
|
||
|
||
核心逻辑：
|
||
A股弱势行业（白酒/医药/通信/汽车/地产）当前动量负、处于下跌趋势：
|
||
死叉（价格跌破 MA20）→ 模拟做空开仓（下跌中赚钱）
|
||
金叉（价格上穿 MA20）→ 模拟做空平仓（趋势反转，落袋）
|
||
与多账户做多形成多空对冲组合：无论涨跌都有收益来源。
|
||
|
||
与做多的对称关系：
|
||
做多: 金叉买入 → 死叉卖出（赚上涨）
|
||
做空: 死叉开空 → 金叉平空（赚下跌）
|
||
|
||
用法:
|
||
python3 stock_short.py status # 全部空头账户状态
|
||
python3 stock_short.py scan # 扫描弱势行业死叉/金叉信号
|
||
python3 stock_short.py report # 汇总报告
|
||
python3 stock_short.py backtest <code> <name> # 单只做空回测
|
||
"""
|
||
import json, os, subprocess, sys, time
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
HOME = Path.home()
|
||
OUTPUT = HOME / ".hermes" / "stock_backtest" / "short_account"
|
||
OUTPUT.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 弱势行业 → 做空标的（回测筛选 2026-08-02，不能只看行业动量！）
|
||
# ⚠️ 教训：行业动量负 ≠ 个股跌。中兴通讯/中天科技（算力概念）做空亏 37%/40%——
|
||
# 必须用做空回测逐只验证，只保留"死叉开空能赚钱"的标的。
|
||
# 回测结果：五粮液+25% 万科+16% 长城+10% 恒瑞+3% 联通+4%
|
||
SHORT_TARGETS = {
|
||
"白酒": {"code": "000858", "name": "五粮液", "capital": 100000},
|
||
"医药": {"code": "600276", "name": "恒瑞医药", "capital": 100000},
|
||
"通信": {"code": "600050", "name": "中国联通", "capital": 100000},
|
||
"汽车": {"code": "601633", "name": "长城汽车", "capital": 100000},
|
||
"地产": {"code": "000002", "name": "万科A", "capital": 100000},
|
||
}
|
||
|
||
# 技术参数（与做多对称）
|
||
MA_PERIOD = 20
|
||
|
||
|
||
def get_url(url, timeout=8, enc="utf-8", retries=1):
|
||
"""腾讯行情（curl subprocess 模式，铁律：urllib 在此目录挂起）"""
|
||
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(code):
|
||
return ("sh" if code.startswith(("6", "5")) else "sz") + code
|
||
|
||
|
||
def get_klines(code, days=60):
|
||
"""获取日K前复权 → [(date, close), ...]"""
|
||
mc = market_code(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("="))
|
||
kl = d.get("data", {}).get(mc, {}).get("qfqday") or d.get("data", {}).get(mc, {}).get("day", [])
|
||
return [(row[0], float(row[2])) for row in kl]
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
def get_quote(code):
|
||
mc = market_code(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_file(industry):
|
||
return OUTPUT / f"short_{industry}.json"
|
||
|
||
|
||
def load_account(industry):
|
||
f = account_file(industry)
|
||
if f.exists():
|
||
return json.load(open(f))
|
||
cfg = SHORT_TARGETS.get(industry, {})
|
||
return {
|
||
"industry": industry,
|
||
"stock": cfg.get("name", ""),
|
||
"code": cfg.get("code", ""),
|
||
"initial_capital": cfg.get("capital", 0),
|
||
"current_capital": cfg.get("capital", 0),
|
||
"short_positions": [], # [{"shares", "open_price", "open_date", "pnl"}]
|
||
"closed_shorts": [], # [{"shares", "open_price", "close_price", "pnl", "open_date", "close_date"}]
|
||
"last_signal": "空仓",
|
||
"last_signal_date": "",
|
||
"stats": {"total_shorts": 0, "winning_shorts": 0, "losing_shorts": 0},
|
||
"created": datetime.now().strftime("%Y-%m-%d"),
|
||
}
|
||
|
||
|
||
def save_account(acct):
|
||
with open(account_file(acct["industry"]), "w") as f:
|
||
json.dump(acct, f, ensure_ascii=False, indent=2)
|
||
|
||
|
||
def compute_ma20_signal(klines):
|
||
"""计算 MA20 金叉/死叉信号（做空视角）：
|
||
返回: "SHORT_OPEN"(死叉，跌破MA20) / "SHORT_CLOSE"(金叉，上穿MA20) / "HOLD"
|
||
做空对称：死叉开空，金叉平空
|
||
"""
|
||
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 prev_above and not cur_above:
|
||
return "SHORT_OPEN", cur_close, cur_ma20 # 跌破 → 开空
|
||
if not prev_above and cur_above:
|
||
return "SHORT_CLOSE", cur_close, cur_ma20 # 上穿 → 平空
|
||
return "HOLD", cur_close, cur_ma20
|
||
|
||
|
||
def open_short(acct, price, dry_run=False):
|
||
"""做空开仓：卖出借入的股票，期望价格下跌后买回"""
|
||
alloc = 1.0 # 做空标的都是弱势行业，用全仓做空额度（模拟）
|
||
shares = int(acct["current_capital"] * alloc // price)
|
||
if shares <= 0:
|
||
return "SKIP", "资金不足"
|
||
if dry_run:
|
||
return "SHORT_OPEN", f"[DRY-RUN] 做空开仓 {shares}股 @{price:.2f}"
|
||
acct["short_positions"].append({
|
||
"shares": shares,
|
||
"open_price": price,
|
||
"open_date": datetime.now().strftime("%Y-%m-%d"),
|
||
"pnl": 0,
|
||
})
|
||
acct["last_signal"] = f"做空开仓 {shares}股 @{price:.2f}"
|
||
acct["last_signal_date"] = datetime.now().strftime("%Y-%m-%d")
|
||
save_account(acct)
|
||
return "SHORT_OPEN", f"做空开仓 {shares}股 @{price:.2f}（跌到平仓线盈利）"
|
||
|
||
|
||
def close_short(acct, price, dry_run=False):
|
||
"""做空平仓：买回股票归还，盈亏 = (开仓价 - 平仓价) * 股数"""
|
||
if not acct["short_positions"]:
|
||
return "SKIP", "无空头持仓"
|
||
pos = acct["short_positions"][0]
|
||
shares = pos["shares"]
|
||
pnl = (pos["open_price"] - price) * shares
|
||
pnl_pct = pnl / (pos["open_price"] * shares) * 100
|
||
if dry_run:
|
||
return "SHORT_CLOSE", f"[DRY-RUN] 做空平仓 {shares}股 盈亏{pnl:+,.0f} ({pnl_pct:+.2f}%)"
|
||
acct["closed_shorts"].append({
|
||
"shares": shares,
|
||
"open_price": pos["open_price"],
|
||
"close_price": price,
|
||
"pnl": round(pnl, 2),
|
||
"open_date": pos["open_date"],
|
||
"close_date": datetime.now().strftime("%Y-%m-%d"),
|
||
})
|
||
acct["current_capital"] += pnl
|
||
acct["short_positions"] = []
|
||
acct["stats"]["total_shorts"] += 1
|
||
if pnl >= 0:
|
||
acct["stats"]["winning_shorts"] += 1
|
||
else:
|
||
acct["stats"]["losing_shorts"] += 1
|
||
acct["last_signal"] = f"做空平仓 盈亏{pnl:+,.0f} ({pnl_pct:+.2f}%)"
|
||
acct["last_signal_date"] = datetime.now().strftime("%Y-%m-%d")
|
||
save_account(acct)
|
||
return "SHORT_CLOSE", f"做空平仓 {shares}股 盈亏{pnl:+,.0f} ({pnl_pct:+.2f}%)"
|
||
|
||
|
||
def execute_short(sig, dry_run=False):
|
||
"""信号执行（cron 用）: sig = {"industry": "白酒", "signal": "SHORT_OPEN"/"SHORT_CLOSE", "close": price}"""
|
||
industry = sig.get("industry")
|
||
if industry not in SHORT_TARGETS:
|
||
return "SKIP", f"非做空标的行业: {industry}"
|
||
acct = load_account(industry)
|
||
signal = sig.get("signal")
|
||
price = sig.get("close", 0)
|
||
if signal == "SHORT_OPEN":
|
||
return open_short(acct, price, dry_run)
|
||
if signal == "SHORT_CLOSE":
|
||
return close_short(acct, price, dry_run)
|
||
return "SKIP", f"未知信号 {signal}"
|
||
|
||
|
||
def scan_signals():
|
||
"""扫描全部做空标的 → 信号列表"""
|
||
results = []
|
||
for industry, cfg in SHORT_TARGETS.items():
|
||
klines = get_klines(cfg["code"])
|
||
if not klines:
|
||
results.append({"industry": industry, "signal": "DATA_ERR", "close": 0})
|
||
continue
|
||
sig, close, ma20 = compute_ma20_signal(klines)
|
||
dev = (close - ma20) / ma20 * 100 if ma20 else 0
|
||
results.append({
|
||
"industry": industry,
|
||
"code": cfg["code"],
|
||
"name": cfg["name"],
|
||
"signal": sig,
|
||
"close": close,
|
||
"ma20": ma20,
|
||
"dev": dev,
|
||
})
|
||
return results
|
||
|
||
|
||
def backtest(code, name, industry):
|
||
"""做空策略回测：遍历历史K线，死叉开空/金叉平空"""
|
||
klines = get_klines(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_short = False
|
||
open_price = 0
|
||
open_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 prev_above and not cur_above and not in_short:
|
||
in_short = True
|
||
open_price = cur_close
|
||
open_date = dates[i]
|
||
elif not prev_above and cur_above and in_short:
|
||
pnl = (open_price - cur_close) * int(100000 / open_price)
|
||
trades.append({"open": open_date, "close": dates[i], "open_p": open_price, "close_p": cur_close, "pnl": pnl})
|
||
capital += pnl
|
||
in_short = False
|
||
total_pnl = capital - 100000
|
||
win = sum(1 for t in trades if t["pnl"] > 0)
|
||
return {
|
||
"code": code, "name": name, "industry": industry,
|
||
"strategy_return": total_pnl / 100000 * 100,
|
||
"total_shorts": len(trades),
|
||
"win_rate": win / len(trades) * 100 if trades else 0,
|
||
"total_pnl": total_pnl,
|
||
"in_short": in_short,
|
||
"last_trades": trades[-5:],
|
||
}
|
||
|
||
|
||
def cmd_status():
|
||
print(f"小唯模拟做空 — {datetime.now().strftime('%Y-%m-%d %H:%M')}")
|
||
print("=" * 50)
|
||
total_cap = 0
|
||
for industry in SHORT_TARGETS:
|
||
acct = load_account(industry)
|
||
total_cap += acct["current_capital"]
|
||
pos = acct["short_positions"]
|
||
if pos:
|
||
p = pos[0]
|
||
print(f"【{industry}】🔴做空中 {acct['stock']} {p['shares']}股 @{p['open_price']:.2f} 开仓{p['open_date']}")
|
||
else:
|
||
print(f"【{industry}】⚪空仓 {acct['stock']}({acct['code']}) 资金{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 industry in SHORT_TARGETS:
|
||
acct = load_account(industry)
|
||
total_cap += acct["current_capital"]
|
||
total_win += acct["stats"]["winning_shorts"]
|
||
total_lose += acct["stats"]["losing_shorts"]
|
||
mark = "🔴" if acct["short_positions"] else "⚪"
|
||
print(f"{mark} {industry}: {acct['stock']} 资产{acct['current_capital']:,.0f} "
|
||
f"(做空{acct['stats']['total_shorts']} 胜{acct['stats']['winning_shorts']} 负{acct['stats']['losing_shorts']})")
|
||
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 = {"SHORT_OPEN": "🔴开空", "SHORT_CLOSE": "🟢平空", "HOLD": "⚪持有", "DATA_ERR": "❌数据"}.get(s["signal"], "?")
|
||
print(f"{mark} {s['industry']}: {s.get('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:
|
||
# 参数顺序: backtest <code> <name> [industry]
|
||
industry = sys.argv[4] if len(sys.argv) > 4 else "自定义"
|
||
r = backtest(sys.argv[2], sys.argv[3], industry)
|
||
if r:
|
||
print(f"做空回测 {r['name']}({r['code']}) [{r['industry']}]")
|
||
print(f" 策略收益: {r['strategy_return']:+.2f}% | 做空{r['total_shorts']}次 | 胜率{r['win_rate']:.0f}% | 总盈亏{r['total_pnl']:+,.0f}")
|
||
if r["last_trades"]:
|
||
print(" 最近交易:")
|
||
for t in r["last_trades"]:
|
||
print(f" {t['open']} 空@{t['open_p']:.2f} → {t['close']} 平@{t['close_p']:.2f} 盈亏{t['pnl']:+,.0f}")
|
||
else:
|
||
print("数据不足")
|
||
else:
|
||
print(__doc__)
|