309 lines
11 KiB
Python
309 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
小唯每日组合信号 — 全市场MA20扫描
|
||
==================================
|
||
每日扫描 ⭐关注 股票，给出综合评分和最优选择
|
||
|
||
覆盖: 五粮液/贵州茅台/泸州老窖/洋河股份/招商银行/中国平安/平安银行
|
||
|
||
用法:
|
||
python3 stock_portfolio.py # 打印组合信号
|
||
python3 stock_portfolio.py --push # 推送飞书
|
||
"""
|
||
|
||
import json, sys, urllib.request, re
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
OUTPUT = Path.home() / ".hermes" / "stock_backtest"
|
||
OUTPUT.mkdir(exist_ok=True)
|
||
FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/446db983-e392-4d2c-bfb8-f9060e5df3ad"
|
||
|
||
|
||
def load_backtest_stats(code):
|
||
"""读取回测结果，用于金叉时的置信度参考"""
|
||
result_file = OUTPUT / f"ma20_result_{code}.json"
|
||
if not result_file.exists():
|
||
return None
|
||
try:
|
||
import ast
|
||
with open(result_file) as f:
|
||
d = json.load(f)
|
||
return d
|
||
except Exception:
|
||
return None
|
||
|
||
# MA20参数
|
||
MA_PERIOD = 20
|
||
|
||
# ⭐关注股票列表
|
||
WATCHED_STOCKS = [
|
||
("000858", "五粮液", "白酒"),
|
||
("600519", "贵州茅台", "白酒"),
|
||
("000568", "泸州老窖", "白酒"),
|
||
("002304", "洋河股份", "白酒"),
|
||
("600036", "招商银行", "银行"),
|
||
("601318", "中国平安", "保险"),
|
||
("000001", "平安银行", "银行"),
|
||
]
|
||
|
||
|
||
def get_url(url, timeout=8):
|
||
"""统一用curl，避免urllib在hermes/scripts目录下异常"""
|
||
import subprocess, shlex, os
|
||
env = dict(os.environ)
|
||
for k in ["http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY"]:
|
||
env.pop(k, None)
|
||
cmd = f"curl -s --max-time {timeout} --compressed {shlex.quote(url)}"
|
||
try:
|
||
r = subprocess.run(cmd, shell=True, capture_output=True, timeout=timeout + 2, env=env)
|
||
return r.stdout.decode("utf-8", errors="ignore")
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def get_url_gbk(url, timeout=8):
|
||
"""GBK编码的请求（腾讯行情）"""
|
||
import subprocess, shlex, os
|
||
env = dict(os.environ)
|
||
for k in ["http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY"]:
|
||
env.pop(k, None)
|
||
cmd = f"curl -s --max-time {timeout} --compressed {shlex.quote(url)}"
|
||
try:
|
||
r = subprocess.run(cmd, shell=True, capture_output=True, timeout=timeout + 2, env=env)
|
||
return r.stdout.decode("gbk", errors="ignore")
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def get_kline_data(code, min_needed=60):
|
||
"""获取足够的日K线数据（前复权）"""
|
||
import subprocess, shlex, os, re, json
|
||
from datetime import timedelta
|
||
market = "sz" if code.startswith(("00", "30")) else "sh"
|
||
full = f"{market}{code}"
|
||
# 实际交易日只有2/3，请求足够多的数据确保有min_needed个
|
||
end_d = datetime.now()
|
||
start_d = end_d - timedelta(days=int(min_needed * 1.8))
|
||
start_str = start_d.strftime("%Y-%m-%d")
|
||
end_str = end_d.strftime("%Y-%m-%d")
|
||
# 请求足够多的条数
|
||
url = (f"https://web.ifzq.gtimg.cn/appstock/app/fqkline/get"
|
||
f"?_var=kline_dayqfq¶m={full},day,{start_str},{end_str},300,qfq")
|
||
env = dict(os.environ)
|
||
for k in ["http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY"]:
|
||
env.pop(k, None)
|
||
cmd = f"curl -s --max-time 8 --compressed {shlex.quote(url)}"
|
||
try:
|
||
r = subprocess.run(cmd, shell=True, capture_output=True, timeout=10, env=env)
|
||
text = r.stdout.decode("utf-8", errors="ignore")
|
||
except Exception:
|
||
return []
|
||
if not text or len(text) < 50:
|
||
print(f"[DEBUG] {code} empty response, len={len(text)}")
|
||
return []
|
||
try:
|
||
json_str = re.sub(r"^[^=]+=", "", text, count=1)
|
||
data = json.loads(json_str)
|
||
raw = data.get("data", {}).get(full, {}).get("qfqday", [])
|
||
if not raw:
|
||
raw = data.get("data", {}).get(full, {}).get("day", [])
|
||
if raw:
|
||
raw = sorted(raw, key=lambda x: x[0])[-min_needed:]
|
||
return raw
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
def validate_klines(raw, min_needed=25):
|
||
"""验证K线数据质量，返回有效数据或空列表"""
|
||
if not raw or len(raw) < min_needed:
|
||
return []
|
||
# 检查日期连续性（允许周末/节假日跳跃）
|
||
dates = [r[0] for r in raw]
|
||
prices = [float(r[2]) for r in raw if len(r) > 2]
|
||
if len(prices) < min_needed:
|
||
return []
|
||
# 检查价格合理性：过滤价格为0或异常值
|
||
valid = [r for r in raw if len(r) > 5 and float(r[2]) > 0]
|
||
return valid[-min_needed:] if len(valid) >= min_needed else []
|
||
|
||
|
||
def analyze_ma20(code, name, industry):
|
||
"""分析单只股票的MA20状态 — 含数据质量验证"""
|
||
raw_klines = get_kline_data(code, MA_PERIOD + 10)
|
||
klines = validate_klines(raw_klines, MA_PERIOD + 2)
|
||
if len(klines) < MA_PERIOD + 2:
|
||
return None
|
||
|
||
try:
|
||
closes = [float(k[2]) for k in klines] # index 2 = close
|
||
ma20 = sum(closes[-MA_PERIOD:]) / MA_PERIOD
|
||
current_price = closes[-1]
|
||
prev_price = closes[-2]
|
||
|
||
above_ma = current_price > ma20
|
||
golden_cross = prev_price <= ma20 < current_price # 今天刚上穿
|
||
dead_cross = prev_price >= ma20 > current_price # 今天刚下穿
|
||
|
||
chg_pct = (current_price - closes[0]) / closes[0] * 100 if closes else 0
|
||
|
||
return {
|
||
"code": code,
|
||
"name": name,
|
||
"industry": industry,
|
||
"price": current_price,
|
||
"ma20": ma20,
|
||
"above_ma": above_ma,
|
||
"golden_cross": golden_cross,
|
||
"dead_cross": dead_cross,
|
||
"chg_pct": chg_pct,
|
||
"diff_pct": (current_price - ma20) / ma20 * 100,
|
||
}
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def get_macro():
|
||
"""宏观指标：上证 + 原油 + 沪深300（大盘风险代理）"""
|
||
result = {}
|
||
# 上证
|
||
text = get_url_gbk("https://qt.gtimg.cn/q=sh000001")
|
||
parts = text.split("~")
|
||
if len(parts) > 5:
|
||
try:
|
||
p = float(parts[3])
|
||
y = float(parts[4])
|
||
result["上证"] = {"price": p, "change": (p-y)/y*100}
|
||
except:
|
||
pass
|
||
# 原油
|
||
text = get_url_gbk("https://qt.gtimg.cn/q=hf_OIL")
|
||
parts = text.split("~")
|
||
if len(parts) > 5 and parts[3]:
|
||
try:
|
||
p = float(parts[3])
|
||
y = float(parts[4])
|
||
result["原油"] = {"price": p, "change": p-y}
|
||
except:
|
||
pass
|
||
# 沪深300（大盘风险代理）
|
||
text = get_url_gbk("https://qt.gtimg.cn/q=sh510300")
|
||
parts = text.split("~")
|
||
if len(parts) > 5:
|
||
try:
|
||
p = float(parts[3])
|
||
y = float(parts[4])
|
||
chg = (p-y)/y*100
|
||
result["沪深300"] = {"price": p, "change": chg,
|
||
"level": "强势(>0)" if chg > 0 else "弱势(<0)"}
|
||
except:
|
||
pass
|
||
# 离岸人民币（新浪 ifzq 备用，失败则跳过）
|
||
try:
|
||
url = "https://web.ifzq.gtimg.cn/appstock/app/fqkline/get?_var=kline_dayqfq¶m=usdcny,day,2026-07-01,2026-07-12,5,qfq"
|
||
text = get_url(url)
|
||
# 解析失败则跳过，不影响其他指标
|
||
except Exception:
|
||
pass
|
||
return result
|
||
|
||
|
||
def build_portfolio_report(push=False):
|
||
today = datetime.now().strftime("%Y-%m-%d")
|
||
macro = get_macro()
|
||
|
||
print(f"\n{'='*60}")
|
||
print(f"小唯股票组合信号 {today}")
|
||
print(f"{'='*60}")
|
||
|
||
# 宏观
|
||
print("\n【宏观】")
|
||
for name, d in macro.items():
|
||
e = "📈" if d["change"] > 0 else "📉"
|
||
print(f" {e} {name}: {d['price']:.2f} ({d['change']:+.2f})")
|
||
|
||
# 个股扫描
|
||
print(f"\n【MA20扫描】MA周期={MA_PERIOD}日")
|
||
print(f"{'代码':<8} {'名称':<8} {'行业':<6} {'价格':>8} {'MA20':>8} {'偏离':>7} {'信号'}")
|
||
print("-" * 65)
|
||
|
||
signals = [] # 有信号的股票
|
||
for code, name, industry in WATCHED_STOCKS:
|
||
r = analyze_ma20(code, name, industry)
|
||
if not r:
|
||
print(f"{code} {name}: 数据不足")
|
||
continue
|
||
|
||
above = "✅在MA20上方" if r["above_ma"] else "❌在MA20下方"
|
||
diff = r["diff_pct"]
|
||
|
||
signal = "持仓"
|
||
signal_icon = "🟢"
|
||
if r["golden_cross"]:
|
||
signal = "⭐金叉买入"
|
||
signal_icon = "🟡"
|
||
elif r["dead_cross"]:
|
||
signal = "🔴死叉卖出"
|
||
signal_icon = "🔴"
|
||
elif not r["above_ma"]:
|
||
signal = "空仓"
|
||
signal_icon = "⚪"
|
||
|
||
print(f" {code} {name:<6} {industry:<5} {r['price']:>8.2f} {r['ma20']:>8.2f} {diff:>+6.1f}% {signal_icon}{signal}")
|
||
|
||
if r["golden_cross"]:
|
||
signals.append({"code": code, "name": name, "price": r["price"], "ma20": r["ma20"], "diff": diff})
|
||
|
||
# 总结
|
||
print(f"\n{'='*60}")
|
||
if signals:
|
||
print(f"⭐ 今日出现MA20金叉 ({len(signals)}只):")
|
||
for s in signals:
|
||
stats = load_backtest_stats(s["code"])
|
||
stat_line = ""
|
||
if stats:
|
||
stat_line = f" | 历史胜率{stats.get('win_rate',0):.0f}% α{stats.get('alpha',0):+.1f}% 最大回撤{stats.get('max_drawdown',0):.0f}%"
|
||
print(f" → {s['name']}({s['code']}) 价格{s['price']:.2f} MA20={s['ma20']:.2f} 偏离{s['diff']:+.1f}%{stat_line}")
|
||
else:
|
||
print("今日无金叉信号，所有股票在MA20下方，继续空仓等待。")
|
||
|
||
print(f"\n小唯股票投研 · 组合扫描")
|
||
|
||
# 飞书推送
|
||
if push:
|
||
msg = f"📊 股票组合信号 {today}\n\n"
|
||
for name, d in macro.items():
|
||
e = "📈" if d["change"] > 0 else "📉"
|
||
msg += f"{e} {name}: {d['price']:.2f} ({d['change']:+.2f}%)\n"
|
||
msg += f"\nMA20扫描 ({MA_PERIOD}日):\n"
|
||
for code, name, industry in WATCHED_STOCKS:
|
||
r = analyze_ma20(code, name, industry)
|
||
if not r:
|
||
continue
|
||
diff = r["diff_pct"]
|
||
if r["golden_cross"]:
|
||
stats = load_backtest_stats(code)
|
||
stat_line = ""
|
||
if stats:
|
||
stat_line = f"\n 📊 MA20策略: 历史胜率{stats.get('win_rate',0):.0f}% | α{stats.get('alpha',0):+.1f}% | 最大回撤{stats.get('max_drawdown',0):.0f}%"
|
||
msg += f"🟡 {name}({code}) 金叉! 价格{r['price']:.2f} 偏离MA20 {diff:+.1f}%{stat_line}\n"
|
||
elif r["dead_cross"]:
|
||
msg += f"🔴 {name}({code}) 死叉! 平仓\n"
|
||
if not signals:
|
||
msg += "\n暂无金叉，继续空仓等待。\n"
|
||
msg += f"\n生成: {datetime.now().strftime('%H:%M')}"
|
||
try:
|
||
payload = json.dumps({"msg_type": "text", "content": {"text": msg}}).encode()
|
||
req = urllib.request.Request(FEISHU_WEBHOOK, data=payload,
|
||
headers={"Content-Type": "application/json"})
|
||
urllib.request.urlopen(req, timeout=8)
|
||
print("✅ 已推送飞书")
|
||
except Exception:
|
||
print("⚠️ 飞书推送失败")
|
||
|
||
return signals
|
||
|
||
|
||
if __name__ == "__main__":
|
||
build_portfolio_report(push="--push" in sys.argv) |