234 lines
8.2 KiB
Python
234 lines
8.2 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日均线
|
||
ma20 = float(parts[35]) # 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": {
|
||
"macro": -1, # 宏观承压
|
||
"fundamental": 1, # 基本面稳健
|
||
"technical": -1, # 技术面空头
|
||
"message": 0, # 消息面中性
|
||
"total": -1, # 综合-1
|
||
"decision": "空仓等待，继续积累数据"
|
||
}
|
||
}
|
||
|
||
return contradiction
|
||
|
||
def print_report(c):
|
||
"""输出矛盾分析报告"""
|
||
verdict = c["weekly_verdict"]
|
||
|
||
print("=" * 50)
|
||
print(f"📅 {c['date']} 矛盾分析报告")
|
||
print("=" * 50)
|
||
|
||
print("\n【四维评分】")
|
||
print(f" 宏观面: {verdict['macro']}（系统性压力）")
|
||
print(f" 基本面: {verdict['fundamental']}（稳健）")
|
||
print(f" 技术面: {verdict['technical']}（空头排列）")
|
||
print(f" 消息面: {verdict['message']}（中性）")
|
||
print(f" 综合评分: {verdict['total']} → 决策：{verdict['decision']}")
|
||
|
||
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() |