218 lines
7.3 KiB
Python
218 lines
7.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
小唯股票基本面采集 — 腾讯行情接口（含 PE/PB/市值）
|
||
==================================================
|
||
从 qt.gtimg.cn 实时接口提取基本面指标，供四维评分系统使用。
|
||
腾讯返回 88 字段，关键字段：
|
||
[1] 名称 [3] 现价 [4] 昨收 [5] 今开 [6] 成交量
|
||
[39] PE(TTM) [43] 换手率 [44] PB [45] 总市值(亿)
|
||
[46] 流通市值(亿) [47] 涨停价 [48] 跌停价 [52] 60日涨幅
|
||
[53] 年初至今涨幅 [54] 52周高 [55] 52周低 [69] 总股本
|
||
[70] 流通股本 [71] 每股收益EPS [73] 每股净资产
|
||
[83] 52周区间 [84] 市净率 [85] 市盈率(动)
|
||
|
||
用法:
|
||
python3 stock_fundamental.py 000858 600519 # 批量查询
|
||
"""
|
||
import sys, urllib.request
|
||
from datetime import datetime
|
||
import json
|
||
from pathlib import Path
|
||
|
||
OUTPUT = Path.home() / ".hermes" / "stock_backtest"
|
||
OUTPUT.mkdir(exist_ok=True)
|
||
|
||
STOCK_NAMES = {
|
||
"000858": "五粮液", "600519": "贵州茅台", "000568": "泸州老窖",
|
||
"002304": "洋河股份", "600036": "招商银行", "601318": "中国平安",
|
||
"000001": "平安银行", "300750": "宁德时代", "002594": "比亚迪",
|
||
"002415": "海康威视", "000063": "中兴通讯", "601088": "中国神华",
|
||
"600188": "兖矿能源", "688981": "中芯国际", "600030": "中信证券",
|
||
"000333": "美的集团", "600276": "恒瑞医药", "600941": "中国移动",
|
||
"601633": "长城汽车", "000002": "万科A",
|
||
}
|
||
|
||
|
||
def get_fundamental(code):
|
||
"""获取单只股票基本面指标"""
|
||
mc = f"sh{code}" if code.startswith(("6", "9")) else f"sz{code}"
|
||
try:
|
||
url = f"https://qt.gtimg.cn/q={mc}"
|
||
data = urllib.request.urlopen(url, timeout=8).read().decode("gbk")
|
||
parts = data.split("~")
|
||
if len(parts) < 50:
|
||
return None
|
||
|
||
def f(idx, default=0.0):
|
||
try:
|
||
v = float(parts[idx])
|
||
return v if v == v else default # NaN check
|
||
except (ValueError, IndexError):
|
||
return default
|
||
|
||
return {
|
||
"code": code,
|
||
"name": parts[1].strip() if len(parts) > 1 else code,
|
||
"price": f(3),
|
||
"yesterday": f(4),
|
||
"open": f(5),
|
||
"volume": f(6), # 手
|
||
"pe_ttm": f(39), # 市盈率TTM
|
||
"turnover": f(38), # 换手率
|
||
"pb": f(43) or f(84), # 市净率
|
||
"mktcap": f(44), # 总市值(亿)
|
||
"float_cap": f(45), # 流通市值(亿)
|
||
"high52w": f(53) if f(53) > 0 else f(54), # 52周高
|
||
"low52w": f(55), # 52周低
|
||
"eps": f(71), # 每股收益
|
||
"bvps": f(73), # 每股净资产
|
||
"pe_dynamic": f(85), # 动态市盈率
|
||
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M"),
|
||
}
|
||
except Exception as e:
|
||
print(f" ⚠️ {code} 基本面获取失败: {e}")
|
||
return None
|
||
|
||
|
||
def batch_get_fundamental(codes):
|
||
"""批量获取基本面（一次请求多个代码，腾讯支持逗号分隔）"""
|
||
symbols = []
|
||
for c in codes:
|
||
symbols.append(f"sh{c}" if c.startswith(("6", "9")) else f"sz{c}")
|
||
url = "https://qt.gtimg.cn/q=" + ",".join(symbols)
|
||
try:
|
||
data = urllib.request.urlopen(url, timeout=15).read().decode("gbk")
|
||
except Exception as e:
|
||
print(f" ⚠️ 批量请求失败: {e}")
|
||
return {}
|
||
|
||
result = {}
|
||
for line in data.strip().split(";"):
|
||
line = line.strip()
|
||
if "=" not in line:
|
||
continue
|
||
content = line.split("=", 1)[1].strip().strip('"')
|
||
parts = content.split("~")
|
||
if len(parts) < 50:
|
||
continue
|
||
code = parts[2].strip()
|
||
|
||
def f(idx, default=0.0):
|
||
try:
|
||
v = float(parts[idx])
|
||
return v if v == v else default
|
||
except (ValueError, IndexError):
|
||
return default
|
||
|
||
result[code] = {
|
||
"code": code,
|
||
"name": parts[1].strip(),
|
||
"price": f(3),
|
||
"yesterday": f(4),
|
||
"pe_ttm": f(39),
|
||
"turnover": f(38),
|
||
"pb": f(43) or f(84),
|
||
"mktcap": f(44),
|
||
"float_cap": f(45),
|
||
"high52w": f(53) if f(53) > 0 else f(54),
|
||
"low52w": f(55),
|
||
"eps": f(71),
|
||
"bvps": f(73),
|
||
"pe_dynamic": f(85),
|
||
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M"),
|
||
}
|
||
return result
|
||
|
||
|
||
def score_fundamental(fd):
|
||
"""
|
||
基本面四维评分（用于四维体系）
|
||
PE 评分：行业中性参考，<20 低估，20-40 合理，>40 偏贵（对成长股放宽）
|
||
简化版：PE 低 + PB 低 + 大市值 = 稳健
|
||
返回: (score, reasons)
|
||
score: -1=差 / 0=中性 / +1=强
|
||
"""
|
||
if not fd:
|
||
return (0, ["基本面数据缺失"])
|
||
|
||
reasons = []
|
||
score = 0
|
||
|
||
pe = fd.get("pe_ttm", 0)
|
||
pb = fd.get("pb", 0)
|
||
mktcap = fd.get("mktcap", 0)
|
||
|
||
# PE 判断
|
||
if pe <= 0:
|
||
reasons.append(f"PE={pe:.0f}(亏损或异常)")
|
||
score -= 1
|
||
elif pe < 20:
|
||
reasons.append(f"PE={pe:.1f}(低估)")
|
||
score += 1
|
||
elif pe < 40:
|
||
reasons.append(f"PE={pe:.1f}(合理)")
|
||
else:
|
||
reasons.append(f"PE={pe:.1f}(偏贵)")
|
||
score -= 1
|
||
|
||
# PB 判断
|
||
if pb > 0:
|
||
if pb < 2:
|
||
reasons.append(f"PB={pb:.2f}(低估值)")
|
||
score += 1
|
||
elif pb < 5:
|
||
reasons.append(f"PB={pb:.2f}(合理)")
|
||
else:
|
||
reasons.append(f"PB={pb:.2f}(高估值)")
|
||
score -= 1
|
||
|
||
# 市值判断（大市值稳健）
|
||
if mktcap > 2000:
|
||
reasons.append(f"市值{mktcap:.0f}亿(大盘蓝筹)")
|
||
score += 1
|
||
elif mktcap > 500:
|
||
reasons.append(f"市值{mktcap:.0f}亿(中盘)")
|
||
else:
|
||
reasons.append(f"市值{mktcap:.0f}亿(小盘波动大)")
|
||
score -= 1
|
||
|
||
return (max(-1, min(1, score)), reasons)
|
||
|
||
|
||
def batch_scan(codes=None):
|
||
"""批量扫描关注股票基本面（一次请求全部）"""
|
||
codes = codes or list(STOCK_NAMES.keys())
|
||
results = []
|
||
print("=" * 60)
|
||
print(f"小唯股票基本面扫描 {datetime.now().strftime('%Y-%m-%d %H:%M')}")
|
||
print("=" * 60)
|
||
print(f"{'代码':<8} {'名称':<8} {'现价':>8} {'PE':>6} {'PB':>5} {'市值亿':>8} {'EPS':>6} {'评分'}")
|
||
print("-" * 60)
|
||
|
||
all_fd = batch_get_fundamental(codes)
|
||
for code in codes:
|
||
fd = all_fd.get(code)
|
||
if not fd:
|
||
print(f" ⚠️ {code} 获取失败")
|
||
continue
|
||
sc, reasons = score_fundamental(fd)
|
||
results.append({"fundamental": fd, "score": sc, "reasons": reasons})
|
||
emoji = "🟢" if sc > 0 else ("🔴" if sc < 0 else "⚪")
|
||
print(f"{code:<8} {fd['name']:<8} {fd['price']:>8.2f} {fd['pe_ttm']:>6.1f} "
|
||
f"{fd['pb']:>5.2f} {fd['mktcap']:>8.0f} {fd['eps']:>6.2f} {emoji}{sc:+d}")
|
||
|
||
# 保存
|
||
out = OUTPUT / "fundamental_scan.json"
|
||
with open(out, "w", encoding="utf-8") as f:
|
||
json.dump({"generated": datetime.now().strftime("%Y-%m-%d %H:%M"),
|
||
"results": results}, f, ensure_ascii=False, indent=2)
|
||
print(f"\n📁 已保存: {out}")
|
||
return results
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) > 1:
|
||
batch_scan(sys.argv[1:])
|
||
else:
|
||
batch_scan()
|