386 lines
14 KiB
Python
386 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
小唯股票投研 — 矛盾分析工作流
|
||
===============================
|
||
用马克思主义矛盾分析法指导每周投资复盘。
|
||
|
||
原理：
|
||
1. 实践是检验真理的唯一标准 → 模拟持仓验证判断
|
||
2. 对立统一 → 市场里的多空矛盾
|
||
3. 量变质变 → MA20金叉/死叉是质变信号
|
||
4. 否定之否定 → 每周复盘迭代策略
|
||
|
||
用法：
|
||
python3 stock_contradiction_workflow.py report # 输出本周矛盾分析
|
||
python3 stock_contradiction_workflow.py check # 检查持仓状态（MA20信号）
|
||
python3 stock_contradiction_workflow.py record # 记录本周分析到历史
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import sys
|
||
from datetime import datetime
|
||
|
||
# ============ 配置 ============
|
||
HISTORY_FILE = os.path.expanduser("~/.hermes/stock_backtest/contradiction_history.json")
|
||
OUTPUT_DIR = os.path.expanduser("~/.hermes/stock_backtest/")
|
||
|
||
# ============ 四维评分标准 ============
|
||
# 宏观面：-1=系统性压力 / 0=中性 / +1=友好
|
||
# 基本面：-1=差 / 0=中性 / +1=强
|
||
# 技术面：-1=下跌趋势 / 0=震荡 / +1=上升趋势
|
||
# 消息面：-1=利空 / 0=中性 / +1=利好
|
||
|
||
def load_history():
|
||
"""加载历史矛盾分析记录"""
|
||
if os.path.exists(HISTORY_FILE):
|
||
with open(HISTORY_FILE) as f:
|
||
return json.load(f)
|
||
return {"weeks": [], "positions": {}}
|
||
|
||
def save_history(history):
|
||
"""保存历史记录"""
|
||
os.makedirs(os.path.dirname(HISTORY_FILE), exist_ok=True)
|
||
with open(HISTORY_FILE, "w") as f:
|
||
json.dump(history, f, ensure_ascii=False, indent=2)
|
||
|
||
def get_ma20_signal(code):
|
||
"""
|
||
查 MA20 状态（用腾讯行情 API）
|
||
返回：{"code": "000858", "name": "五粮液", "signal": "golden_cross"|"dead_cross"|"neutral", "price": xxx}
|
||
"""
|
||
import urllib.request
|
||
|
||
# 转换代码格式
|
||
if code.startswith("0") or code.startswith("3"):
|
||
market = "sz"
|
||
else:
|
||
market = "sh"
|
||
|
||
url = f"https://qt.gtimg.cn/q={market}{code}"
|
||
try:
|
||
with urllib.request.urlopen(url, timeout=5) as resp:
|
||
data = resp.read().decode("gbk")
|
||
|
||
parts = data.split("~")
|
||
if len(parts) < 33:
|
||
return None
|
||
|
||
name = parts[1]
|
||
price = float(parts[3]) # 现价
|
||
ma5 = float(parts[33]) # 5日均线
|
||
ma10 = float(parts[34]) # 10日均线
|
||
# 处理20日均线字段：可能是纯数字或"数字/..."格式（腾讯API字段变化）
|
||
ma20_raw = parts[35]
|
||
if '/' in ma20_raw:
|
||
ma20_str = ma20_raw.split('/')[0]
|
||
else:
|
||
ma20_str = ma20_raw
|
||
ma20 = float(ma20_str) # 20日均线
|
||
|
||
# 判断 MA20 信号
|
||
if ma5 > ma20 and price > ma20:
|
||
signal = "golden_cross" # 潜在金叉（5日上穿20日）
|
||
elif ma5 < ma20 and price < ma20:
|
||
signal = "dead_cross" # 死叉（5日下穿20日）
|
||
else:
|
||
signal = "neutral"
|
||
|
||
return {
|
||
"code": code,
|
||
"name": name,
|
||
"signal": signal,
|
||
"price": price,
|
||
"ma5": ma5,
|
||
"ma10": ma10,
|
||
"ma20": ma20
|
||
}
|
||
except Exception as e:
|
||
return {"code": code, "error": str(e)}
|
||
|
||
def analyze_week():
|
||
"""生成本周矛盾分析"""
|
||
history = load_history()
|
||
|
||
# 持仓股票
|
||
positions = history.get("positions", {
|
||
"000858": {"name": "五粮液", "status": "空仓等金叉", "avg_cost": None},
|
||
})
|
||
|
||
# 检查 MA20 信号
|
||
signals = {}
|
||
for code in ["000858", "600519", "000568", "002304", "601318", "600036"]:
|
||
result = get_ma20_signal(code)
|
||
if result:
|
||
signals[code] = result
|
||
|
||
# 矛盾分析框架
|
||
now = datetime.now()
|
||
week_num = now.isocalendar()[1]
|
||
|
||
# 核心矛盾（五粮液）
|
||
wuliangye = signals.get("000858", {})
|
||
|
||
contradiction = {
|
||
"week": f"{now.year}-W{week_num:02d}",
|
||
"date": now.strftime("%Y-%m-%d"),
|
||
"market_contradictions": [
|
||
{
|
||
"contradiction": "宏观经济下行 vs 消费刚需（白酒）",
|
||
"manifestation": "宏观面-1，但五粮液基本面+1",
|
||
"current_phase": "矛盾积累阶段，等待技术面确认方向",
|
||
"action": "空仓等待，观察宏观拐点信号"
|
||
},
|
||
{
|
||
"contradiction": "估值回归 vs 趋势惯性",
|
||
"manifestation": "PE跌至历史低位，但均线仍空头排列",
|
||
"current_phase": "量变质变的临界积累期",
|
||
"action": "等MA20金叉（质变信号）再入场"
|
||
}
|
||
],
|
||
"position_status": positions,
|
||
"ma20_signals": {code: {k: v for k, v in s.items() if k != "error"}
|
||
for code, s in signals.items()},
|
||
"weekly_verdict": build_verdict(signals)
|
||
}
|
||
|
||
return contradiction
|
||
|
||
|
||
def load_fundamental_scan():
|
||
"""读取基本面扫描结果（stock_fundamental.py 生成）"""
|
||
f = os.path.join(OUTPUT_DIR, "fundamental_scan.json")
|
||
if not os.path.exists(f):
|
||
return None
|
||
try:
|
||
with open(f) as fp:
|
||
return json.load(fp)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def load_macro_score():
|
||
"""读取宏观评分结果（stock_macro.py 生成，--json 模式）"""
|
||
f = os.path.join(OUTPUT_DIR, "macro_score.json")
|
||
if not os.path.exists(f):
|
||
return None
|
||
try:
|
||
with open(f) as fp:
|
||
return json.load(fp)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def load_sentiment_scan():
|
||
"""读取消息面情感扫描结果（stock_sentiment.py 生成）"""
|
||
f = os.path.join(OUTPUT_DIR, "sentiment_scan.json")
|
||
if not os.path.exists(f):
|
||
return None
|
||
try:
|
||
with open(f) as fp:
|
||
return json.load(fp)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def build_verdict(signals):
|
||
"""根据真实数据计算四维评分（2026-08-02 增强：宏观接入真实数据）"""
|
||
now = datetime.now()
|
||
|
||
# 宏观面：真实数据（stock_macro.py 生成）
|
||
macro = 0
|
||
macro_reasons = []
|
||
macro_data = load_macro_score()
|
||
if macro_data:
|
||
macro = macro_data.get("macro", 0)
|
||
macro_reasons = macro_data.get("reasons", [])
|
||
else:
|
||
macro = -1 # 回退：无数据时保持保守判断
|
||
macro_reasons = ["宏观数据缺失，保守-1"]
|
||
|
||
# 基本面：取关注股票 PE/PB 平均，判断整体估值
|
||
fundamental = 0
|
||
fund_reasons = []
|
||
scan = load_fundamental_scan()
|
||
if scan and scan.get("results"):
|
||
# 只取白酒+金融核心关注股
|
||
core_codes = {"000858", "600519", "000568", "002304", "600036", "601318", "000001"}
|
||
scores = []
|
||
for r in scan["results"]:
|
||
fd = r.get("fundamental", {})
|
||
if fd.get("code") in core_codes:
|
||
scores.append(r.get("score", 0))
|
||
if scores:
|
||
avg_f = sum(scores) / len(scores)
|
||
fundamental = 1 if avg_f > 0.3 else (-1 if avg_f < -0.3 else 0)
|
||
fund_reasons.append(f"核心股基本面均分{avg_f:+.1f}")
|
||
else:
|
||
fundamental = 1 # 回退：无数据时保持原判断
|
||
|
||
# 技术面：MA20 信号判断
|
||
technical = 0
|
||
above_count = sum(1 for s in signals.values() if s.get("signal") == "golden_cross")
|
||
below_count = sum(1 for s in signals.values() if s.get("signal") == "dead_cross")
|
||
if above_count > below_count:
|
||
technical = 1
|
||
elif below_count > above_count:
|
||
technical = -1
|
||
|
||
# 消息面：真实数据（stock_sentiment.py 生成）
|
||
message = 0
|
||
msg_reasons = []
|
||
senti = load_sentiment_scan()
|
||
if senti:
|
||
raw = senti.get("message_score", 0)
|
||
message = 1 if raw > 0.3 else (-1 if raw < -0.3 else 0)
|
||
if message != 0:
|
||
msg_reasons.append(f"市场消息面{raw:+.1f}")
|
||
|
||
total = macro + fundamental + technical + message
|
||
|
||
# 决策
|
||
if total >= 2:
|
||
decision = "偏多，可关注强势行业金叉"
|
||
elif total <= -2:
|
||
decision = "偏空，空仓等待"
|
||
else:
|
||
decision = "震荡观望，等MA20金叉确认方向"
|
||
|
||
return {
|
||
"macro": macro,
|
||
"fundamental": fundamental,
|
||
"technical": technical,
|
||
"message": message,
|
||
"total": total,
|
||
"decision": decision,
|
||
"macro_reasons": macro_reasons,
|
||
"fundamental_reasons": fund_reasons,
|
||
"message_reasons": msg_reasons,
|
||
}
|
||
|
||
def load_industry_momentum():
|
||
"""读取行业扫描结果（stock_industry_scan.py 生成）"""
|
||
f = os.path.join(OUTPUT_DIR, "industry_scan.json")
|
||
if not os.path.exists(f):
|
||
return None
|
||
try:
|
||
with open(f) as fp:
|
||
return json.load(fp)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def print_report(c):
|
||
"""输出矛盾分析报告"""
|
||
verdict = c["weekly_verdict"]
|
||
|
||
print("=" * 50)
|
||
print(f"📅 {c['date']} 矛盾分析报告")
|
||
print("=" * 50)
|
||
|
||
print("\n【四维评分】")
|
||
macro_line = "（承压）" if verdict['macro'] < 0 else ("（友好）" if verdict['macro'] > 0 else "（中性）")
|
||
print(f" 宏观面: {verdict['macro']}{macro_line}")
|
||
if verdict.get("macro_reasons"):
|
||
print(f" {' '.join(verdict['macro_reasons'][:2])}")
|
||
fund_line = "（稳健）" if verdict['fundamental'] > 0 else ("（偏弱）" if verdict['fundamental'] < 0 else "（中性）")
|
||
print(f" 基本面: {verdict['fundamental']}{fund_line}")
|
||
if verdict.get("fundamental_reasons"):
|
||
print(f" {' '.join(verdict['fundamental_reasons'])}")
|
||
print(f" 技术面: {verdict['technical']}（空头排列）")
|
||
msg_line = "（利好）" if verdict['message'] > 0 else ("（利空）" if verdict['message'] < 0 else "（中性）")
|
||
print(f" 消息面: {verdict['message']}{msg_line}")
|
||
if verdict.get("message_reasons"):
|
||
print(f" {' '.join(verdict['message_reasons'])}")
|
||
print(f" 综合评分: {verdict['total']} → 决策：{verdict['decision']}")
|
||
|
||
# 行业动量版块（2026-08-01 新增）
|
||
scan = load_industry_momentum()
|
||
if scan and scan.get("industries"):
|
||
print("\n【行业动量排名】(学术因子扫描)")
|
||
inds = scan["industries"]
|
||
# 格式：{"avg_mom": {"行业": 值, ...}, "avg_sharpe": {...}}
|
||
try:
|
||
mom_map = inds.get("avg_mom", {})
|
||
sharpe_map = inds.get("avg_sharpe", {})
|
||
ranked = sorted(mom_map.items(), key=lambda kv: kv[1], reverse=True)
|
||
for ind, mom in ranked[:8]:
|
||
sharpe = sharpe_map.get(ind, 0)
|
||
emoji = "🟢" if mom > 0 else "🔴"
|
||
print(f" {emoji} {ind}: 动量{mom:+.1%} Sharpe{sharpe:+.2f}")
|
||
print(f" ... (共{len(mom_map)}行业，弱势拦截: 白酒/医药/通信/汽车/地产)")
|
||
except Exception:
|
||
print(" (数据解析失败)")
|
||
else:
|
||
print("\n【行业动量】无扫描数据，运行 stock_industry_scan.py 生成")
|
||
|
||
print("\n【本周核心矛盾】")
|
||
for i, m in enumerate(c["market_contradictions"], 1):
|
||
print(f"\n 矛盾{i}：{m['contradiction']}")
|
||
print(f" 表现：{m['manifestation']}")
|
||
print(f" 阶段：{m['current_phase']}")
|
||
print(f" 行动：{m['action']}")
|
||
|
||
print("\n【MA20信号监控】")
|
||
if c["ma20_signals"]:
|
||
for code, sig in c["ma20_signals"].items():
|
||
if "error" in sig:
|
||
print(f" {code}: 查询失败")
|
||
else:
|
||
name = sig.get("name", code)
|
||
signal = sig.get("signal", "unknown")
|
||
emoji = {"golden_cross": "🟢金叉", "dead_cross": "🔴死叉", "neutral": "⚪中性"}.get(signal, signal)
|
||
price = sig.get("price", 0)
|
||
ma20 = sig.get("ma20", 0)
|
||
print(f" {name}（{code}）{emoji} 现价:{price:.2f} MA20:{ma20:.2f}")
|
||
else:
|
||
print(" 无数据")
|
||
|
||
print("\n【持仓状态】")
|
||
for code, pos in c["position_status"].items():
|
||
print(f" {pos['name']}（{code}）：{pos['status']}")
|
||
|
||
print("\n" + "=" * 50)
|
||
print("【马克思主义分析框架】")
|
||
print(f" 本质：实践是检验真理的唯一标准")
|
||
print(f" 当前实践：模拟持仓，等MA20金叉验证判断")
|
||
print(f" 矛盾积累：宏观-1+基本面+1，等待技术面确认方向")
|
||
print(f" 量变质变：MA20金叉=质变临界点")
|
||
print("=" * 50)
|
||
|
||
def main():
|
||
cmd = sys.argv[1] if len(sys.argv) > 1 else "report"
|
||
|
||
if cmd == "report":
|
||
c = analyze_week()
|
||
print_report(c)
|
||
|
||
# 保存到历史
|
||
history = load_history()
|
||
history["weeks"].append(c)
|
||
if len(history["weeks"]) > 52: # 保留一年
|
||
history["weeks"] = history["weeks"][-52:]
|
||
save_history(history)
|
||
|
||
elif cmd == "check":
|
||
# 快速检查 MA20 信号
|
||
for code in ["000858", "600519", "000568", "002304"]:
|
||
sig = get_ma20_signal(code)
|
||
if sig and "error" not in sig:
|
||
emoji = {"golden_cross": "🟢", "dead_cross": "🔴", "neutral": "⚪"}.get(sig["signal"], "❓")
|
||
print(f"{emoji} {sig['name']} | 现价:{sig['price']:.2f} MA20:{sig['ma20']:.2f}")
|
||
elif sig:
|
||
print(f"❌ {code}: {sig.get('error', 'unknown')}")
|
||
|
||
elif cmd == "record":
|
||
c = analyze_week()
|
||
history = load_history()
|
||
if c not in history["weeks"]:
|
||
history["weeks"].append(c)
|
||
save_history(history)
|
||
print(f"✅ 已记录 {c['week']} 的矛盾分析")
|
||
else:
|
||
print(f"⚠️ {c['week']} 已记录，跳过")
|
||
|
||
if __name__ == "__main__":
|
||
main() |