163 lines
5.3 KiB
Python
163 lines
5.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
小唯宏观指标自动化 — 真实数据驱动四维评分的宏观维度
|
||
==================================================
|
||
获取上证/沪深300/创业板/原油，计算宏观趋势评分。
|
||
|
||
评分规则：
|
||
上证 20日涨幅 > 0 → +1（大盘向上）
|
||
上证 20日涨幅 < -3% → -1（大盘走弱）
|
||
沪深300 20日涨幅 > 0 → +1
|
||
原油 20日涨幅 > 10% → -1（通胀压力）
|
||
综合：sum 后 clip 到 [-1, 1]
|
||
|
||
用法：
|
||
python3 stock_macro.py # 输出宏观评分
|
||
python3 stock_macro.py --json # JSON 输出（供矛盾周报引用）
|
||
"""
|
||
import json, sys, urllib.request
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
|
||
OUTPUT = Path.home() / ".hermes" / "stock_backtest"
|
||
OUTPUT.mkdir(exist_ok=True)
|
||
|
||
|
||
def get_url(url, timeout=8):
|
||
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):
|
||
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_change(symbol, days=20):
|
||
"""获取指数/品种过去 N 日涨幅"""
|
||
today = datetime.now().strftime("%Y-%m-%d")
|
||
start = (datetime.now() - timedelta(days=days * 2)).strftime("%Y-%m-%d")
|
||
url = (f"https://web.ifzq.gtimg.cn/appstock/app/fqkline/get"
|
||
f"?_var=kline_dayqfq¶m={symbol},day,{start},{today},{days},qfq")
|
||
text = get_url(url)
|
||
if not text or len(text) < 50:
|
||
return None
|
||
try:
|
||
import re
|
||
json_str = re.sub(r"^[^=]+=", "", text, count=1)
|
||
data = json.loads(json_str)
|
||
key = list(data.get("data", {}).keys())
|
||
if not key:
|
||
return None
|
||
raw = data["data"][key[0]].get("qfqday") or data["data"][key[0]].get("day") or []
|
||
if len(raw) < 2:
|
||
return None
|
||
closes = [float(r[2]) for r in raw if len(r) > 2 and float(r[2]) > 0]
|
||
if len(closes) < 2:
|
||
return None
|
||
return (closes[-1] - closes[0]) / closes[0] * 100
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def get_macro_score():
|
||
"""
|
||
计算宏观评分
|
||
返回: {"macro": int, "details": {指标: 涨幅}, "reasons": [...]}
|
||
"""
|
||
# 上证指数
|
||
shanghai = get_kline_change("sh000001", days=20)
|
||
# 沪深300（用 ETF 510300 代理）
|
||
hs300 = get_kline_change("sh510300", days=20)
|
||
# 创业板
|
||
chinext = get_kline_change("sz399006", days=20)
|
||
# 原油
|
||
oil = get_kline_change("hf_OIL", days=20)
|
||
|
||
score = 0
|
||
details = {}
|
||
reasons = []
|
||
|
||
if shanghai is not None:
|
||
details["上证20日"] = f"{shanghai:+.1f}%"
|
||
if shanghai > 0:
|
||
score += 1
|
||
reasons.append(f"上证20日{shanghai:+.1f}% → 大盘向上")
|
||
elif shanghai < -3:
|
||
score -= 1
|
||
reasons.append(f"上证20日{shanghai:+.1f}% → 大盘走弱")
|
||
else:
|
||
reasons.append(f"上证20日{shanghai:+.1f}% → 震荡")
|
||
|
||
if hs300 is not None:
|
||
details["沪深300"] = f"{hs300:+.1f}%"
|
||
if hs300 > 0:
|
||
score += 1
|
||
reasons.append(f"沪深300 {hs300:+.1f}% → 蓝筹强")
|
||
elif hs300 < -3:
|
||
score -= 1
|
||
reasons.append(f"沪深300 {hs300:+.1f}% → 蓝筹弱")
|
||
|
||
if chinext is not None:
|
||
details["创业板"] = f"{chinext:+.1f}%"
|
||
if chinext < -5:
|
||
score -= 1
|
||
reasons.append(f"创业板{chinext:+.1f}% → 成长弱")
|
||
|
||
if oil is not None:
|
||
details["原油20日"] = f"{oil:+.1f}%"
|
||
if oil > 10:
|
||
score -= 1
|
||
reasons.append(f"原油{oil:+.1f}% → 通胀压力")
|
||
elif oil < -10:
|
||
score += 1
|
||
reasons.append(f"原油{oil:+.1f}% → 通缩缓解")
|
||
|
||
# clip 到 [-1, 1]
|
||
macro = max(-1, min(1, score))
|
||
if macro == 0 and not reasons:
|
||
macro = 0
|
||
reasons.append("宏观数据获取不足，中性")
|
||
|
||
return {"macro": macro, "details": details, "reasons": reasons, "date": datetime.now().strftime("%Y-%m-%d")}
|
||
|
||
|
||
def main():
|
||
result = get_macro_score()
|
||
if "--json" in sys.argv:
|
||
out = OUTPUT / "macro_score.json"
|
||
with open(out, "w", encoding="utf-8") as f:
|
||
json.dump(result, f, ensure_ascii=False, indent=2)
|
||
print(json.dumps(result, ensure_ascii=False))
|
||
print(f"📁 已保存: {out}")
|
||
return
|
||
|
||
print("=" * 50)
|
||
print(f"小唯宏观评分 {result['date']}")
|
||
print("=" * 50)
|
||
for k, v in result["details"].items():
|
||
print(f" {k}: {v}")
|
||
print(f"\n宏观评分: {result['macro']:+d}")
|
||
for r in result["reasons"]:
|
||
print(f" • {r}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|