xiaowei-system/scripts/stock_selector.py

529 lines
20 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""
小唯四维选股系统 — 宏观+基本面+技术+消息
========================================
综合四个维度筛选股票
维度说明:
1. 宏观面 — 美联储政策/人民币汇率/地缘风险
2. 基本面 — PE/PB/业绩增速/行业景气度
3. 技术面 — MA20趋势/成交量/均线排列
4. 消息面 — 政策利好/行业新闻/重大事件
评分规则:
- 宏观负面(-1)/中性(0)/正面(+1)
- 基本面低估(+1)/合理(0)/高估(-1)
- 技术面多头(+1)/空头(-1)/震荡(0)
- 消息面利好(+1)/利空(-1)/中性(0)
- 综合评分: 4分以上关注, 6分以上重点
用法:
python3 stock_selector.py scan # 扫描全市场
python3 stock_selector.py analyze <代码> # 分析单只股票
python3 stock_selector.py watchlist # 展示关注列表
"""
import json, sys, urllib.request, time
from datetime import datetime
from pathlib import Path
import numpy as np
import pandas as pd
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 send_feishu(msg):
payload = json.dumps({"msg_type": "text", "content": {"text": msg}}).encode()
req = urllib.request.Request(FEISHU_WEBHOOK, data=payload,
headers={"Content-Type": "application/json"})
try:
with urllib.request.urlopen(req, timeout=10):
pass
except Exception:
pass
# ===================== 宏观面 — 汇率/利率/风险 =====================
def get_macro_score():
"""
宏观面评分:
考虑: 人民币汇率, 美联储政策, 地缘风险
返回: -1(负面) / 0(中性) / +1(正面)
"""
score = 0
reasons = []
# 人民币汇率
try:
url = "https://qt.gtimg.cn/q=usdcnh"
text = urllib.request.urlopen(url, timeout=5).read().decode("gbk")
parts = text.split("~")
if len(parts) > 10:
usdcnh = float(parts[3])
if usdcnh > 7.3:
score -= 1
reasons.append(f"离岸人民币破7.3({usdcnh}) → 外资流出压力 ⬇️")
else:
score += 1
reasons.append(f"人民币相对稳定({usdcnh}) → 外资流出压力小 ⬆️")
except Exception:
reasons.append("人民币汇率获取失败")
# A股风险偏好(用沪深300判断)
try:
url = "https://web.ifzq.gtimg.cn/appstock/app/fqkline/get?_var=kline_dayqfq&param=sh510300,day,2026-07-01,2026-07-12,10,qfq"
text = urllib.request.urlopen(url, timeout=5).read().decode("utf-8")
data = json.loads(text.replace("kline_dayqfq=", "", 1))
rows = data.get("data", {}).get("sh510300", {}).get("qfqday") or []
if len(rows) >= 5:
closes = [float(r[2]) for r in rows]
if closes[-1] > closes[0]:
score += 1
reasons.append("沪深300近期上涨 → 市场风险偏好上升 ⬆️")
else:
score -= 1
reasons.append("沪深300近期下跌 → 市场风险偏好下降 ⬇️")
except Exception:
pass
# 平均
avg_score = max(-1, min(1, score // max(1, len([r for r in reasons if r]))))
return avg_score, reasons
# ===================== 基本面 — PE/PB/业绩 =====================
def get_fundamental_score(code):
"""
基本面评分:
基于估值(PE/PB)和近期趋势
返回: -1(高估) / 0(合理) / +1(低估)
"""
try:
mc = f"sh{code}" if code.startswith("6") else f"sz{code}"
url = f"https://qt.gtimg.cn/q={mc}"
text = urllib.request.urlopen(url, timeout=5).read().decode("gbk")
parts = text.split("~")
if len(parts) < 40:
return 0, [], {}
pe = float(parts[39]) if parts[39] else 0
pb = float(parts[46]) if parts[46] else 0
price = float(parts[3])
yclose = float(parts[4])
score = 0
reasons = []
# PE判断
if 0 < pe < 15:
score += 1
reasons.append(f"PE={pe:.1f}(历史低位) → 估值有支撑 ⬆️")
elif 15 <= pe <= 30:
reasons.append(f"PE={pe:.1f}(合理区间) → 无明显低估/高估")
elif pe > 30:
score -= 1
reasons.append(f"PE={pe:.1f}(历史高位) → 估值偏高风险 ⬇️")
elif pe <= 0:
reasons.append(f"PE={pe}(亏损/无效) → 无法判断")
# PB判断
if 0 < pb < 3:
score += 1
reasons.append(f"PB={pb:.1f}(相对低估) → 净资产有支撑 ⬆️")
elif pb >= 3:
reasons.append(f"PB={pb:.1f}(相对高估)")
# 近期价格位置
try:
url2 = f"https://web.ifzq.gtimg.cn/appstock/app/fqkline/get?_var=kline_dayqfq&param={mc},day,2026-01-01,2026-07-12,200,qfq"
text2 = urllib.request.urlopen(url2, timeout=5).read().decode("utf-8")
data2 = json.loads(text2.replace("kline_dayqfq=", "", 1))
hist = data2.get("data", {}).get(mc, {}).get("qfqday") or []
if len(hist) >= 60:
highs = [float(r[3]) for r in hist]
low = min(highs)
high = max(highs)
pos = (price - low) / (high - low) if high > low else 0.5
if pos < 0.2:
score += 1
reasons.append(f"价格处于近半年低位({pos*100:.0f}%) → 相对安全边际 ⬆️")
elif pos > 0.8:
score -= 1
reasons.append(f"价格处于近半年高位({pos*100:.0f}%) → 追高风险 ⬇️")
except Exception:
pass
avg_score = max(-1, min(1, score))
info = {"pe": pe, "pb": pb, "price": price, "yclose": yclose}
return avg_score, reasons, info
except Exception as e:
return 0, [f"基本面数据获取失败({e})"], {}
# ===================== 技术面 — MA20趋势 =====================
def get_technical_score(code):
"""
技术面评分:
基于MA20均线状态
返回: -1(空头) / 0(震荡) / +1(多头)
"""
try:
mc = f"sh{code}" if code.startswith("6") else f"sz{code}"
end = datetime.now().strftime("%Y-%m-%d")
url = f"https://web.ifzq.gtimg.cn/appstock/app/fqkline/get?_var=kline_dayqfq&param={mc},day,2025-07-01,{end},500,qfq"
text = urllib.request.urlopen(url, timeout=8).read().decode("utf-8")
data = json.loads(text.replace("kline_dayqfq=", "", 1))
rows = data.get("data", {}).get(mc, {}).get("qfqday") or []
if len(rows) < 30:
return 0, ["数据不足"], {}
closes = [float(r[2]) for r in rows]
dates = [r[0] for r in rows]
closes_s = pd.Series(closes)
ma20 = closes_s.rolling(20).mean()
ma60 = closes_s.rolling(60).mean()
ma120 = closes_s.rolling(120).mean()
last_close = closes[-1]
last_ma20 = ma20.iloc[-1]
last_ma60 = ma60.iloc[-1]
last_ma120 = ma120.iloc[-1]
# 均线排列
if last_ma20 > last_ma60 > last_ma120:
ma_arrangement = "多头排列"
t_score = 1
elif last_ma20 < last_ma60 < last_ma120:
ma_arrangement = "空头排列"
t_score = -1
else:
ma_arrangement = "均线混乱"
t_score = 0
# 价格与MA20关系
pct_above = (last_close - last_ma20) / last_ma20 * 100
# 近期趋势
trend_20d = (closes[-1] - closes[-20]) / closes[-20] * 100 if len(closes) >= 20 else 0
reasons = [
f"{ma_arrangement}(价格{'' if last_close>last_ma20 else ''}{pct_above:+.1f}%vsMA20)",
f"近20日涨跌: {trend_20d:+.1f}%",
]
info = {
"price": last_close,
"ma20": last_ma20,
"ma60": last_ma60,
"ma120": last_ma120,
"ma_arrangement": ma_arrangement,
"pct_above_ma20": pct_above,
"trend_20d": trend_20d,
}
return t_score, reasons, info
except Exception as e:
return 0, [f"技术面数据获取失败({e})"], {}
# ===================== 消息面 — 行业/政策 =====================
def get_stock_news(code, name, max_news=3):
"""获取个股最新新闻(东方财富)"""
try:
url = f"https://np-anotice-stock.eastmoney.com/api/security/ann?sr=-1&page_size={max_news}&page_index=1&ann_type=SZA&stock_list={code}"
import subprocess, shlex, os, json
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 6 --compressed {shlex.quote(url)}"
r = subprocess.run(cmd, shell=True, capture_output=True, timeout=8, env=env)
data = json.loads(r.stdout.decode("utf-8", errors="ignore"))
notices = data.get("data", {}).get("list", [])
news = []
for n in notices[:max_news]:
title = n.get("title", "")[:50]
notice_date = n.get("notice_date", "")[:10]
if title:
news.append(f"· {notice_date} {title}")
return news if news else [f"{max_news}日无重大公告"]
except Exception:
return ["新闻获取失败"]
def get_sector_news(industry, max_news=2):
"""获取行业相关新闻 — 回退到指数情绪代理行业API不可用时"""
try:
# 先尝试东财快讯
keyword_map = {"白酒": "白酒", "银行": "银行", "保险": "保险", "新能源": "新能源"}
keyword = keyword_map.get(industry, industry) or industry or ""
base = "https://search-api-web.eastmoney.com/search/jsonp"
param = ('{"uid":"","keyword":"' + keyword + '","type":["cmsArticleListNew"],'
'"client":"web","clientVersion":"curr","clientType":"web",'
'"param":{"cmsArticleListNew":{"pageIndex":1,"pageSize":' + str(max_news) + '}}}')
url = base + "?param=" + param
env = dict(os.environ)
for k in ["http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY"]:
env.pop(k, None)
cmd = ["curl", "-s", "--max-time", "6",
"-H", "Referer: https://so.eastmoney.com/",
"-H", "User-Agent: Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 Chrome/120 Safari/537.36",
url]
r = subprocess.run(cmd, capture_output=True, timeout=8, env=env)
raw = r.stdout.decode("utf-8", errors="ignore")
m = re.search(r'\(\[.*\]\)', raw, re.DOTALL)
if m:
items = json.loads(m.group(1))
news = [f"· {it.get('title','')[:40]}" for it in items[:max_news] if it.get('title')]
if news:
return news
except Exception:
pass
# 回退:使用指数涨跌作为行业情绪代理
try:
sector_codes = {"白酒": "sh000858", "银行": "sh000001", "保险": "sh601318", "新能源": "sz399808"}
code = sector_codes.get(industry, "sh000001")
mc = code[:2] + code[2:]
url = f"https://qt.gtimg.cn/q={mc}"
text = get_url_gbk(url)
parts = text.split("~")
if len(parts) > 5:
p = float(parts[3])
y = float(parts[4])
chg = (p-y)/y*100
sentiment = "📈 行业指数上涨" if chg > 0 else "📉 行业指数下跌"
return [f"· {sentiment} ({chg:+.2f}%) — {parts[1]}"]
except Exception:
pass
return ["· 行业新闻获取失败(回退到静态规则)"]
def get_sentiment_score_from_news(code, name="", industry=""):
"""消息面评分 — 基于真实新闻 + 静态规则回退"""
import time
news = get_stock_news(code, name)
sector_news = get_sector_news(industry) if industry else []
score = 0
reasons = []
# 真实新闻分析
all_news = news + sector_news
for item in all_news:
text = item.lower()
if any(k in text for k in ["违规", "调查", "处罚", "减持", "业绩预亏", "暴雷"]):
score -= 1
reasons.append(f"利空公告: {item[:30]}")
elif any(k in text for k in ["回购", "增持", "业绩预增", "中标", "合作", "突破"]):
score += 1
reasons.append(f"利好公告: {item[:30]}")
# 静态规则回退(无新闻时)
if not reasons:
if name in ["五粮液", "贵州茅台", "泸州老窖", "洋河股份"]:
score -= 1; reasons.append("白酒消费降级压制 ⬇️")
score += 1; reasons.append("估值低位有支撑 ⬆️")
elif name in ["平安银行", "招商银行"]:
score += 1; reasons.append("高股息防御 ⬆️")
score += 1; reasons.append("低估值稳健 ⬆️")
elif "宁德" in name:
score -= 1; reasons.append("新能源产能过剩担忧 ⬇️")
score += 1; reasons.append("行业龙头壁垒 ⬆️")
else:
reasons.append("消息面无明显驱动")
avg_score = max(-1, min(1, score))
return avg_score, reasons, news, sector_news
# ===================== 综合评分 =====================
def analyze_stock(code, name=""):
"""对单只股票进行四维评分"""
print(f"\n{'='*60}")
print(f"四维选股分析: {name}({code})")
print(f"{'='*60}")
# 1. 宏观面
m_score, m_reasons = get_macro_score()
print(f"[宏观面] 评分: {m_score:+d}")
for r in m_reasons:
print(f" - {r}")
# 2. 基本面
f_score, f_reasons, f_info = get_fundamental_score(code)
print(f"[基本面] 评分: {f_score:+d}")
for r in f_reasons:
print(f" - {r}")
# 3. 技术面
t_score, t_reasons, t_info = get_technical_score(code)
print(f"[技术面] 评分: {t_score:+d}")
for r in t_reasons:
print(f" - {r}")
# 4. 消息面
industry_map = {"000858": "白酒", "600519": "白酒", "000568": "白酒", "002304": "白酒",
"600036": "银行", "601318": "保险", "000001": "银行", "300750": "新能源", "510300": ""}
industry = industry_map.get(code, "")
s_score, s_reasons, stock_news, sector_news = get_sentiment_score_from_news(code, name, industry)
print(f"[消息面] 评分: {s_score:+d}")
for r in s_reasons:
print(f" - {r}")
if stock_news:
for n in stock_news[:2]:
print(f" 📰 {n}")
if sector_news:
for n in sector_news[:1]:
print(f" 📢 {n}")
# ===========================================================
# 组合评分逻辑(修正简单相加的缺陷)
# ===========================================================
# 规则1: 技术面空头 + 基本面低估 = 买入机会(逆向)
# 规则2: 技术面空头 + 基本面高估 = 危险加倍(双杀)
# 规则3: 宏观负面时,技术面多头无法持续
# 规则4: 综合评分考虑方向匹配
f_strong = f_score >= 1 # 基本面好(低估)
f_weak = f_score <= -1 # 基本面差(高估)
t_bear = t_score <= -1 # 技术面空头(超跌)
t_bull = t_score >= 1 # 技术面多头
m_bad = m_score <= -1 # 宏观差
# 逆向机会: 技术面空头+基本面好 = 低估买入机会
if t_bear and f_strong:
adjusted_total = 2 # 视为关注机会
combo_reason = "逆向机会: 技术超跌+基本面低估"
# 双杀: 技术差+基本面也差
elif t_bear and f_weak:
adjusted_total = -2
combo_reason = "双杀风险: 技术超跌+基本面高估"
# 技术多头+基本面好 + 宏观不差
elif t_bull and f_strong and not m_bad:
adjusted_total = 3
combo_reason = "共振: 技术+基本面+宏观同向"
# 强势股(宏观好时)
elif t_bull and m_score >= 0:
adjusted_total = 2
combo_reason = "技术多头,宏观中性支撑"
# 常规计算
else:
adjusted_total = m_score + f_score + t_score + s_score
combo_reason = "常规评分"
total = adjusted_total
max_score = 4
print(f"\n组合逻辑: {combo_reason}")
print(f"\n{'='*60}")
print(f"综合评分: {total:+d} / {max_score} ({(total/max_score*100):.0f}%)")
if total >= 6:
verdict = "⭐⭐⭐ 重点关注 — 四维共振,强烈看多"
elif total >= 4:
verdict = "⭐⭐ 关注 — 多维度偏多"
elif total >= 2:
verdict = "⭐ 关注 — 逆向机会或温和看多"
elif total >= 0:
verdict = "⚠️ 观察 — 方向不明"
else:
verdict = "❌ 不碰 — 双杀风险"
print(f"结论: {verdict}")
print(f"{'='*60}")
# 保存
result = {
"code": code, "name": name, "date": datetime.now().strftime("%Y-%m-%d %H:%M"),
"scores": {"macro": m_score, "fundamental": f_score, "technical": t_score, "sentiment": s_score},
"total": total,
"verdict": verdict,
"macro_reasons": m_reasons,
"fundamental_reasons": f_reasons,
"fundamental_info": f_info,
"technical_reasons": t_reasons,
"technical_info": t_info,
"sentiment_reasons": s_reasons,
}
result_file = OUTPUT / f"四维_{code}.json"
with open(result_file, "w") as f:
json.dump(result, f, ensure_ascii=False, indent=2)
print(f"数据存: {result_file}")
return result
def scan_watchlist():
"""扫描关注列表"""
stocks = [
("000858", "五粮液"),
("600519", "贵州茅台"),
("000001", "平安银行"),
("300750", "宁德时代"),
("510300", "沪深300ETF"),
]
print(f"\n{'='*70}")
print(f"{'股票':<12} {'宏观':>5} {'基本面':>6} {'技术面':>6} {'消息面':>6} {'综合':>5} {'结论'}")
print(f"{'-'*70}")
results = []
for code, name in stocks:
r = analyze_stock(code, name)
results.append(r)
sc = r["scores"]
ts = r["total"]
print(f"{name:<12} {sc['macro']:>+4} {sc['fundamental']:>+5} {sc['technical']:>+5} {sc['sentiment']:>+5} {ts:>+4} {r['verdict'].split(' ')[0]}")
time.sleep(0.5) # 避免请求过快
print(f"{'='*70}")
# 汇总
print(f"\n📊 宏观面: {'负面 ⬇️' if results[0]['scores']['macro'] < 0 else '正面 ⬆️' if results[0]['scores']['macro'] > 0 else '中性'}")
good = [r for r in results if r["total"] >= 2]
print(f"可选股票({len(good)}只): {[r['name'] for r in good]}")
msg = f"""📊 四维选股扫描
宏观面: {'负面 ⬇️' if results[0]['scores']['macro'] < 0 else '正面 ⬆️'}
| 股票 | 宏观 | 基本面 | 技术 | 消息 | 综合 |
|------|------|--------|------|------|------|
"""
for r in results:
sc = r["scores"]
msg += f"| {r['name']} | {sc['macro']:>+2} | {sc['fundamental']:>+2} | {sc['technical']:>+2} | {sc['sentiment']:>+2} | **{r['total']:>+2}** |\n"
good = [r for r in results if r["total"] >= 2]
msg += f"\n可选股票({len(good)}): "
msg += " / ".join([r["name"] for r in good]) if good else ""
msg += f"\n\n生成: {datetime.now().strftime('%Y-%m-%d %H:%M')}"
send_feishu(msg)
if __name__ == "__main__":
if "--scan" in sys.argv or "scan" in sys.argv:
scan_watchlist()
elif "--watchlist" in sys.argv:
scan_watchlist()
elif len(sys.argv) >= 3 and sys.argv[1] == "analyze":
code = sys.argv[2]
name = sys.argv[3] if len(sys.argv) > 3 else code
analyze_stock(code, name)
elif len(sys.argv) >= 2 and sys.argv[1] not in ["--scan", "--watchlist"]:
code = sys.argv[1]
name = sys.argv[2] if len(sys.argv) > 2 else code
analyze_stock(code, name)
else:
print("用法:")
print(" python3 stock_selector.py --scan # 扫描关注列表")
print(" python3 stock_selector.py analyze <代码> [名称] # 分析单只")
print(" python3 stock_selector.py 000858 五粮液 # 同上简写")