@clawhub-gold3bear-944fda614c
美股行情与舆情监控工具。当用户询问「美股怎么样」「纳指」「标普」「道指」「美股大盘」「今晚美股」「US股」「美股行情」「美股期货」「NQ」「ES」时使用。支持Yahoo Finance获取实时行情，以及Google News RSS和X/Twitter舆情监控。
---
name: us-stock-radar
description: 美股行情与舆情监控工具。当用户询问「美股怎么样」「纳指」「标普」「道指」「美股大盘」「今晚美股」「US股」「美股行情」「美股期货」「NQ」「ES」时使用。支持Yahoo Finance获取实时行情，以及Google News RSS和X/Twitter舆情监控。
---
# 美股雷达 (US-Stock Radar)
## 数据源总览
| 数据源 | 用途 | 稳定性 |
|--------|------|--------|
| Yahoo Finance | 主要指数（SPY/QQQ/DIA/IWM）实时行情 | ⭐⭐⭐ |
| Yahoo Finance | 个股行情（NVDA/AAPL/MSFT/TSLA等） | ⭐⭐⭐ |
| TradingView 嵌入页 | NQ/ES 期货实时图表（浏览器截图） | ⭐⭐ |
| Google News RSS | 美股突发新闻 | ⭐⭐⭐ |
## 实时行情查询
### NY Fed 宏观利率 API
```python
import requests
def get_macro_rates():
"""NY Fed 官方利率（SOFR/EFFR/OBFR/TGCR/BGCR）"""
url = "https://markets.newyorkfed.org/api/rates/all/latest.json"
headers = {"User-Agent": "Mozilla/5.0", "Accept": "application/json"}
r = requests.get(url, headers=headers, timeout=10)
return r.json()["refRates"]
# 返回: SOFR=3.64%, EFFR=3.64%, OBFR=3.64%, TGCR=3.62%, BGCR=3.62%
```
### FRED 国债收益率
```python
def get_treasury_yields():
"""10Y 和 2Y 国债收益率（无需 API key）"""
for sid, name in [("DGS10", "10Y"), ("DGS2", "2Y")]:
url = f"https://fred.stlouisfed.org/graph/fredgraph.csv?id={sid}&vintage_date=2026-04-24"
r = requests.get(url, timeout=10)
last = r.text.strip().split('\n')[-1] # 格式: "2026-04-22,4.30"
```
### Yahoo Finance（主要指数）
```python
import requests
US_INDICES = {
"^GSPC": "标普500",
"^DJI": "道琼斯",
"^IXIC": "纳斯达克",
"^VIX": "VIX恐慌指数",
"NQ=F": "纳斯达克期货(NQ)",
"ES=F": "标普期货(ES)",
"CL=F": "WTI原油",
"GC=F": "黄金",
"SI=F": "白银",
}
def get_us_indices():
"""批量获取美股指数 + 期货 + 大宗商品"""
symbols = ",".join(US_INDICES.keys())
url = f"https://query1.finance.yahoo.com/v7/finance/quote?symbols={symbols}"
headers = {"User-Agent": "Mozilla/5.0"}
r = requests.get(url, headers=headers, timeout=10)
results = r.json()["quoteResponse"]["result"]
out = {}
for item in results:
sym = item["symbol"]
name = US_INDICES.get(sym, sym)
price = item.get("regularMarketPrice", 0)
prev = item.get("regularMarketPreviousClose", 0)
chg = item.get("regularMarketChange", 0)
pct = item.get("regularMarketChangePercent", 0)
arrow = "🔴" if chg > 0 else "🟢" if chg < 0 else "⚪"
out[sym] = f"{arrow} {name}: {price} {chg:+.2f}({pct:+.2f}%)"
return out
```
### Yahoo Finance（个股行情）
```python
import requests
MAJOR_STOCKS = {
"NVDA": "英伟达",
"AAPL": "苹果",
"MSFT": "微软",
"GOOGL": "谷歌",
"AMZN": "亚马逊",
"META": "Meta",
"TSLA": "特斯拉",
"AMD": "AMD",
"NFLX": "Netflix",
"CRM": "Salesforce",
}
def get_us_stocks(symbols):
"""获取美股个股行情，支持多代码"""
if isinstance(symbols, str):
symbols = [symbols]
sym_str = ",".join([s.upper() for s in symbols])
url = f"https://query1.finance.yahoo.com/v7/finance/quote?symbols={sym_str}"
headers = {"User-Agent": "Mozilla/5.0"}
r = requests.get(url, headers=headers, timeout=10)
results = r.json()["quoteResponse"]["result"]
for item in results:
sym = item["symbol"]
name = item.get("shortName", sym)
price = item.get("regularMarketPrice", 0)
prev = item.get("regularMarketPreviousClose", 0)
chg = item.get("regularMarketChange", 0)
pct = item.get("regularMarketChangePercent", 0)
arrow = "🔴" if chg > 0 else "🟢" if chg < 0 else "⚪"
print(f"{arrow} {name}({sym}): {price} {chg:+.2f}({pct:+.2f}%)")
```
## 主流代码速查
| 股票/指数 | 代码 | 板块 |
|-----------|------|------|
| 标普500 ETF | SPY | 大盘 |
| 纳指100 ETF | QQQ | 科技 |
| 道指 ETF | DIA | 蓝筹 |
| 小盘股 | IWM | 风险偏好 |
| 英伟达 | NVDA | AI/芯片 |
| 苹果 | AAPL | 科技 |
| 特斯拉 | TSLA | 新能源 |
| AMD | AMD | 芯片 |
| 谷歌 | GOOGL | 科技/AI |
| 亚马逊 | AMZN | 电商/云 |
| Meta | META | 社交/AI |
| 微软 | MSFT | 科技/云 |
## 舆情监控
### Google News Live（突发新闻）
```
https://news.google.com/rss/search?q=US+stock+market+when:1h
https://news.google.com/rss/search?q=Nasdaq+S%26P+500+when:1h
https://news.google.com/rss/search?q=Treasury+yield+Fed+when:1h
https://news.google.com/rss/search?q=Nvidia+AI+stock+when:1h
```
### X/Twitter 美股舆情
使用 browser 工具访问 @bearfrom2077：
```
https://x.com/search?q=%24NVDA+%24TSLA+stock&f=live
https://x.com/search?q=nasdaq+fed+rate&f=live
https://x.com/search?q=S%26P+500+earnings&f=live
```
**核心关键词：**
- `$NVDA` / `$TSLA` / `$AMD` — 个股讨论
- `S&P 500` / `Nasdaq` — 大盘
- `Fed rate` / `Treasury` — 宏观
- `CPI` / `jobs report` — 数据发布
## 情报解读框架
| 指标 | 阈值 | 信号 |
|------|------|------|
| VIX | > 20 | 恐慌加剧 |
| VIX | < 15 | 乐观 |
| 纳指 vs 标普 | 纳指强 > 1% | 科技主线 |
| NQ期货 | 盘前大跌 > 1% | A股/港股承压 |
| 黄金 | 突破 2000 | 避险情绪 |
| 10年美债 | 突破 4.5% | 股市压力 |
| SOFR vs Fed Rate | 低于目标区间下限 | 流动性充裕 |
| 10Y - 2Y 利差 | 倒挂加深 | 经济衰退预警 |
| 10Y - 2Y 利差 | 利差扩大 | 衰退风险缓解 |
**分析顺序：**
1. SOFR/EFFR — 基准利率，了解美联储立场
2. 10Y/2Y 国债收益率 — 利率走廊和经济预期
3. 10Y-2Y 利差 — 衰退概率
4. VIX — 市场情绪温度计
5. NQ/ES 期货 — 盘前大盘方向
6. SPY/QQQ/DIA — 三大指数
7. 科技巨头（NVDA/AAPL/MSFT）— 主线
8. 黄金/原油 — 宏观背景
9. 给出综合判断
## Cron 配置建议
| 频率 | 内容 | 适用场景 |
|------|------|----------|
| 每15分钟 | NQ + ES 期货 | A股开盘前参考 |
| 每30分钟 | SPY + QQQ + VIX | 盘中监控 |
| 每小时 | 科技巨头 + 黄金原油 | 宏观背景 |
| 有问才查 | 个股行情 | 被动触发 |
## 快速查询命令
```bash
cd C:\Users\gold3\.openclaw\workspace\skills\us-stock-radar\scripts
# 美股主要指数+期货+大宗商品
python us_index.py
# 个股行情（传入股票代码）
python us_quote.py NVDA TSLA AAPL
# 科技巨头组合
python tech_giants.py
# 美股仪表盘（指数+巨头）
python dashboard.py
```
FILE:scripts/dashboard.py
"""
美股雷达仪表盘：一键汇总指数 + 期货 + 科技巨头
用法: python dashboard.py
"""
import subprocess
import sys
import os
base = os.path.dirname(os.path.abspath(__file__))
print("=" * 55)
print(" 🦞 美股雷达仪表盘")
print("=" * 55)
print("\n>>> 主要指数 + 期货 + 大宗商品")
subprocess.run([sys.executable, os.path.join(base, "us_index.py")])
print("\n>>> 科技巨头 (NVDA/AAPL/MSFT/GOOGL/AMZN/META/TSLA/AMD)")
subprocess.run([sys.executable, os.path.join(base, "us_quote.py"),
"NVDA", "AAPL", "MSFT", "GOOGL", "AMZN", "META", "TSLA", "AMD"])
FILE:scripts/tech_giants.py
"""
美股科技巨头组合
NVDA / AAPL / MSFT / GOOGL / AMZN / META / TSLA / AMD
"""
import subprocess
import sys
import os
base = os.path.dirname(os.path.abspath(__file__))
if __name__ == "__main__":
tech_stocks = ["NVDA", "AAPL", "MSFT", "GOOGL", "AMZN", "META", "TSLA", "AMD"]
print("=== 苹果+微软+谷歌+Meta+英伟达+特斯拉+AMD ===")
subprocess.run([sys.executable, os.path.join(base, "us_quote.py")] + tech_stocks)
FILE:scripts/us_index.py
"""
美股主要指数 + 期货 + 大宗商品 + 宏观利率实时行情
三数据源：
1. Yahoo Finance query1（主）
2. Yahoo Finance query2（备用域名）
3. Finviz HTML 解析（兜底）
"""
import requests
from datetime import datetime, timezone
import concurrent.futures
SYMBOLS = {
# 指数
"^GSPC": "标普500",
"^DJI": "道琼斯",
"^IXIC": "纳斯达克",
"^VIX": "VIX恐慌指数",
# 期货
"NQ=F": "纳指期货(NQ)",
"ES=F": "标普期货(ES)",
"YM=F": "道指期货(YM)",
# 大宗商品
"GC=F": "黄金",
"CL=F": "WTI原油",
"SI=F": "白银",
}
NYFED_RATES_URL = "https://markets.newyorkfed.org/api/rates/all/latest.json"
FRED_URL_TMPL = "https://fred.stlouisfed.org/graph/fredgraph.csv?id={sid}&vintage_date={date}"
# Yahoo Finance 域名池（轮流尝试，绕过单域名限流）
YAHOO_HOSTS = ["query1.finance.yahoo.com", "query2.finance.yahoo.com"]
def _get_quote_from_host(symbol, host):
url = f"https://{host}/v8/finance/chart/{symbol}?interval=1d&range=1d"
headers = {"User-Agent": "Mozilla/5.0"}
r = requests.get(url, headers=headers, timeout=10)
r.raise_for_status()
d = r.json()
meta = d["chart"]["result"][0]["meta"]
price = meta.get("regularMarketPrice")
prev = meta.get("chartPreviousClose") or meta.get("previousClose")
if price is None or prev in (None, 0):
return None
chg = price - prev
pct = chg / prev * 100
return {
"price": price, "chg": chg, "pct": pct,
"high": meta.get("regularMarketDayHigh"),
"low": meta.get("regularMarketDayLow"),
}
def get_quote(symbol):
"""从多个 Yahoo Finance 域名依次尝试，失败则返回 None"""
for host in YAHOO_HOSTS:
try:
result = _get_quote_from_host(symbol, host)
if result:
return result
except Exception:
continue
return None
def get_all_quotes():
results = {}
for sym, name in SYMBOLS.items():
q = get_quote(sym)
if q:
arrow = "🟢" if q["chg"] > 0 else "🔴" if q["chg"] < 0 else "⚪"
results[sym] = {
"name": name, "price": q["price"],
"chg": q["chg"], "pct": q["pct"], "arrow": arrow
}
# Yahoo 全挂时：尝试 Finviz HTML 解析兜底
if not results:
results = _get_finviz_fallback()
return results
def _get_finviz_fallback():
"""
Finviz HTML 解析兜底（Yahoo Finance 全挂时调用）
Finviz 无需 API key，直接抓取指数页面
"""
INDEX_MAP = {
"^GSPC": ("标普500", "S&P 500"),
"^DJI": ("道琼斯", "Dow Jones"),
"^IXIC": ("纳斯达克", "NASDAQ 100"),
"^VIX": ("VIX恐慌指数", "CBOE Volatility Index"),
"NQ=F": ("纳指期货(NQ)", "Nasdaq 100 Futures"),
"ES=F": ("标普期货(ES)", "S&P 500 Futures"),
"GC=F": ("黄金", "Gold"),
"CL=F": ("WTI原油", "Crude Oil"),
"SI=F": ("白银", "Silver"),
}
# Finviz 个股/指数页面
FINVIZ_URL = "https://finviz.com/indices.ashx"
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36",
"Referer": "https://www.finviz.com/",
}
try:
r = requests.get(FINVIZ_URL, headers=headers, timeout=15)
r.raise_for_status()
from bs4 import BeautifulSoup
soup = BeautifulSoup(r.text, "lxml")
results = {}
rows = soup.select("table.screener_table tr")
for row in rows:
cols = row.find_all("td")
if len(cols) < 8:
continue
# Finviz indices 表格：第0列=名称，第2列=涨跌%，第3列=价格
name_text = cols[0].get_text(strip=True)
chg_text = cols[2].get_text(strip=True).replace("%", "")
price_text = cols[3].get_text(strip=True).replace(",", "")
for y_sym, (cn_name, en_name) in INDEX_MAP.items():
if en_name.lower() in name_text.lower() or cn_name in name_text:
try:
pct = float(chg_text)
price = float(price_text)
chg = price * pct / 100
arrow = "🟢" if pct > 0 else "🔴" if pct < 0 else "⚪"
results[y_sym] = {
"name": cn_name, "price": price,
"chg": chg, "pct": pct, "arrow": arrow
}
except ValueError:
continue
break
return results
except Exception:
return {}
def get_macro_rates():
"""获取 NY Fed 宏观利率（SOFR/EFFR/OBFR 等）"""
headers = {"User-Agent": "Mozilla/5.0", "Accept": "application/json"}
try:
r = requests.get(NYFED_RATES_URL, headers=headers, timeout=10)
r.raise_for_status()
data = r.json().get("refRates", [])
if data:
results = {}
labels = {
"SOFR": "SOFR(担保隔夜融资利率)",
"EFFR": "EFFR(有效联邦基金利率)",
"OBFR": "OBFR(银行隔夜融资利率)",
"TGCR": "TGCR(三方一般抵押利率)",
"BGCR": "BGCR(广泛一般抵押利率)",
}
for item in data:
t = item.get("type", "")
eff_date = item.get("effectiveDate", "")
rate = item.get("percentRate") or item.get("average30day")
if rate and t in labels:
try:
results[t] = {"label": labels[t], "rate": float(rate), "date": eff_date}
except (TypeError, ValueError):
continue
return results
except (requests.RequestException, ValueError):
pass
return {}
def get_treasury_yields():
"""获取 FRED 10Y 和 2Y 国债收益率"""
today = datetime.now(timezone.utc).date().isoformat()
results = {}
for sid, label in [("DGS10", "10Y国债收益率"), ("DGS2", "2Y国债收益率")]:
url = FRED_URL_TMPL.format(sid=sid, date=today)
try:
r = requests.get(url, timeout=10)
r.raise_for_status()
lines = [line for line in r.text.strip().split("\n") if line]
for row in reversed(lines[1:]):
last = row.split(",")
if len(last) == 2 and last[1] not in (".", ""):
results[sid] = {"label": label, "rate": float(last[1]), "date": last[0]}
break
except (requests.RequestException, ValueError):
pass
return results
def print_results():
indices = ["^GSPC", "^DJI", "^IXIC", "^VIX"]
futures = ["NQ=F", "ES=F", "YM=F"]
commodities = ["GC=F", "CL=F", "SI=F"]
print("\n=== 📊 美股主要指数 ===")
for sym in indices:
r = QUOTES.get(sym)
if r:
print(f"{r['arrow']} {r['name']:10s} {r['price']:>10.2f} {r['chg']:>+8.2f}({r['pct']:>+6.2f}%)")
print("\n=== 📈 期货 ===")
for sym in futures:
r = QUOTES.get(sym)
if r:
print(f"{r['arrow']} {r['name']:10s} {r['price']:>10.2f} {r['chg']:>+8.2f}({r['pct']:>+6.2f}%)")
print("\n=== 🛢️ 大宗商品 ===")
for sym in commodities:
r = QUOTES.get(sym)
if r:
print(f"{r['arrow']} {r['name']:10s} {r['price']:>10.2f} {r['chg']:>+8.2f}({r['pct']:>+6.2f}%)")
# 宏观利率
print("\n=== 🏛️ 宏观利率 ===")
rates = MACRO.get("rates", {})
if rates:
for t, v in rates.items():
print(f" 📌 {v['label']}: {v['rate']:.3f}% ( {v['date']})")
else:
print(" (NY Fed rates unavailable)")
# 国债收益率
treasuries = MACRO.get("treasuries", {})
if treasuries:
for t, v in treasuries.items():
print(f" 📌 {v['label']}: {v['rate']:.3f}% ( {v['date']})")
else:
print(" (Treasury yields unavailable)")
if __name__ == "__main__":
print("正在获取美股数据...")
QUOTES = get_all_quotes()
MACRO = {"rates": get_macro_rates(), "treasuries": get_treasury_yields()}
print_results()
FILE:scripts/us_quote.py
"""
美股个股实时行情
用法: python us_quote.py NVDA TSLA AAPL
"""
import requests
import sys
MAJOR_STOCKS = {
"NVDA": "英伟达",
"AAPL": "苹果",
"MSFT": "微软",
"GOOGL": "谷歌",
"AMZN": "亚马逊",
"META": "Meta",
"TSLA": "特斯拉",
"AMD": "AMD",
"NFLX": "Netflix",
"CRM": "Salesforce",
"AVGO": "博通",
"ORCL": "甲骨文",
"COIN": "Coinbase",
}
def get_stock_quote(symbol):
"""获取个股行情"""
sym = symbol.upper()
url = f"https://query2.finance.yahoo.com/v8/finance/chart/{sym}?interval=1d&range=1d"
headers = {"User-Agent": "Mozilla/5.0"}
r = requests.get(url, headers=headers, timeout=10)
if r.status_code != 200:
return None
d = r.json()
try:
meta = d["chart"]["result"][0]["meta"]
price = meta.get("regularMarketPrice")
prev = meta.get("chartPreviousClose") or meta.get("previousClose")
if price is None or prev is None:
return None
chg = price - prev
pct = chg / prev * 100
return {"symbol": sym, "name": MAJOR_STOCKS.get(sym, sym),
"price": price, "chg": chg, "pct": pct}
except (KeyError, IndexError):
return None
if __name__ == "__main__":
if len(sys.argv) < 2:
# 默认显示科技巨头
symbols = ["NVDA", "AAPL", "MSFT", "GOOGL", "AMZN", "META", "TSLA"]
else:
symbols = sys.argv[1:]
print("\n=== 🏛️ 美股个股 ===")
for sym in symbols:
q = get_stock_quote(sym)
if q:
arrow = "🟢" if q["chg"] > 0 else "🔴" if q["chg"] < 0 else "⚪"
name = q["name"]
print(f"{arrow} {name:12s}({q['symbol']:6s}) {q['price']:>10.2f} {q['chg']:>+8.2f}({q['pct']:>+6.2f}%)")
else:
print(f"⚪ {sym}: 获取失败")
期货与大宗商品行情监控工具。当用户询问「原油」「黄金」「白银」「天然气」「铜」「期货行情」「大宗商品」「WTI」「布伦特」「金银比」「油金比」时使用。使用 Yahoo Finance chart API 获取实时期货数据。
---
name: futures-radar
description: 期货与大宗商品行情监控工具。当用户询问「原油」「黄金」「白银」「天然气」「铜」「期货行情」「大宗商品」「WTI」「布伦特」「金银比」「油金比」时使用。使用 Yahoo Finance chart API 获取实时期货数据。
---
# 期货雷达 (Futures Radar)
## 覆盖品种
| 类别 | 品种 | 代码 | 单位 |
|------|------|------|------|
| **能源** | WTI原油 | CL=F | 桶 |
| | 布伦特原油 | BZ=F | 桶 |
| | 天然气 | NG=F | 百万英热 |
| **贵金属** | 黄金 | GC=F | 盎司 |
| | 白银 | SI=F | 盎司 |
| | 铂金 | PL=F | 盎司 |
| | 钯金 | PA=F | 盎司 |
| **基本金属** | 铜 | HG=F | 磅 |
| **农产品** | 玉米 | ZC=F | 蒲式耳 |
| | 大豆 | ZS=F | 蒲式耳 |
| | 小麦 | ZW=F | 蒲式耳 |
| **国债期货** | 10年国债 | ZN=F | 面值100 |
| | 30年国债 | ZB=F | 面值100 |
## 数据获取
```python
import requests
def get_quote(symbol):
"""获取期货行情（Yahoo Finance chart API）"""
url = f"https://query2.finance.yahoo.com/v8/finance/chart/{symbol}?interval=1d&range=1d"
headers = {"User-Agent": "Mozilla/5.0"}
r = requests.get(url, headers=headers, timeout=10)
meta = r.json()["chart"]["result"][0]["meta"]
price = meta["regularMarketPrice"]
prev = meta["chartPreviousClose"]
chg = price - prev
pct = chg / prev * 100
return {"price": price, "chg": chg, "pct": pct,
"high": meta["regularMarketDayHigh"],
"low": meta["regularMarketDayLow"]}
```
## 关键比价与信号
| 比价 | 公式 | 含义 |
|------|------|------|
| 金银比 | GC/F ÷ SI/F | >80 银相对便宜（均值），<50 金便宜 |
| 油金比 | CL/F ÷ (GC/F÷100) | >2.5 通胀预期强，<1.5 避险情绪 |
| 金铜比 | GC/F ÷ HG/F | 经济预期（金强→衰退/避险，铜强→复苏） |
## 情报解读框架
| 品种 | 关键信号 |
|------|----------|
| WTI原油 > 100 | 地缘风险/供应紧张，通胀压力 |
| 黄金 > 2000 | 避险情绪浓厚 |
| 白银 < 22 | 工业需求疲软 |
| 天然气 > 4 | 能源危机/冬季需求 |
| 10年国债期货 > 114 | 降息预期增强（债券涨） |
| 10年国债期货 < 110 | 利率上升预期（债券跌） |
| 钯金 > 2000 | 汽车行业需求强（芯片/电动车替代中） |
**分析顺序：**
1. 黄金 + 白银（避险/通胀主线）
2. WTI + 布伦特（地缘/供给）
3. 金银比（相对价值）
4. 国债期货（利率预期→影响所有资产）
5. 天然气 + 铜（实体经济需求）
## 快速查询命令
```bash
cd C:\Users\gold3\.openclaw\workspace\skills\futures-radar\scripts
# 查询全部期货品种
python futures_spot.py
```
FILE:scripts/futures_spot.py
"""
期货雷达 - WTI原油/黄金/白银/天然气/铜 等大宗商品实时行情
使用 Yahoo Finance chart API
"""
import requests
SYMBOLS = {
# 能源
"CL=F": ("WTI原油", "NYMEX WTI 原油期货", "桶"),
"BZ=F": ("布伦特原油", "ICE 布伦特原油期货", "桶"),
"NG=F": ("天然气", "NYMEX 天然气期货", "百万英热"),
# 贵金属
"GC=F": ("黄金", "COMEX 黄金期货", "盎司"),
"SI=F": ("白银", "COMEX 白银期货", "盎司"),
"PL=F": ("铂金", "NYMEX 铂金期货", "盎司"),
"PA=F": ("钯金", "NYMEX 钯金期货", "盎司"),
# 基本金属
"HG=F": ("铜", "COMEX 铜期货", "磅"),
# 农产品
"ZC=F": ("玉米", "CBOT 玉米期货", "蒲式耳"),
"ZS=F": ("大豆", "CBOT 大豆期货", "蒲式耳"),
"ZW=F": ("小麦", "CBOT 小麦期货", "蒲式耳"),
# 美国国债期货（利率预期）
"ZN=F": ("10年国债期货", "CBOT 10年期国债期货", "面值100"),
"ZB=F": ("30年国债期货", "CBOT 30年期国债期货", "面值100"),
# 宏观指数
"DX-Y.NYB": ("DXY美元指数", "ICE 美元指数", "点"),
}
CATEGORIES = {
"能源": ["CL=F", "BZ=F", "NG=F"],
"贵金属": ["GC=F", "SI=F", "PL=F", "PA=F"],
"基本金属": ["HG=F"],
"农产品": ["ZC=F", "ZS=F", "ZW=F"],
"国债": ["ZN=F", "ZB=F"],
"宏观": ["DX-Y.NYB"],
}
def get_quote(symbol):
url = f"https://query2.finance.yahoo.com/v8/finance/chart/{symbol}?interval=1d&range=1d"
headers = {"User-Agent": "Mozilla/5.0"}
try:
r = requests.get(url, headers=headers, timeout=15)
if r.status_code != 200:
return None
d = r.json()
meta = d["chart"]["result"][0]["meta"]
price = meta.get("regularMarketPrice")
prev = meta.get("chartPreviousClose") or meta.get("previousClose")
high = meta.get("regularMarketDayHigh")
low = meta.get("regularMarketDayLow")
if price is None or prev is None or prev == 0:
return None
chg = price - prev
pct = chg / prev * 100
return {"price": price, "prev": prev, "chg": chg, "pct": pct, "high": high, "low": low}
except (requests.RequestException, ValueError, KeyError, IndexError, TypeError):
return None
def fmt_contract(q, sym, info):
name, _, unit_name = info
arrow = "🔴" if q["chg"] > 0 else "🟢" if q["chg"] < 0 else "⚪"
price_prefix = "" if sym == "DX-Y.NYB" else "$"
high = "--" if q["high"] is None else q["high"]
low = "--" if q["low"] is None else q["low"]
return (f"{arrow} {name:8s} {price_prefix}{q['price']:>10.2f} "
f"{q['chg']:>+8.2f}({q['pct']:>+6.2f}%) "
f"高:{high} 低:{low} {unit_name}")
def main():
print("\n" + "=" * 65)
print(" 🛢️ 期货雷达 — 大宗商品实时行情")
print("=" * 65)
all_quotes = {}
for sym in SYMBOLS:
q = get_quote(sym)
if q:
all_quotes[sym] = q
for cat, syms in CATEGORIES.items():
has_data = any(s in all_quotes for s in syms)
if not has_data:
continue
print(f"\n【{cat}】")
for sym in syms:
if sym not in all_quotes:
continue
info = SYMBOLS[sym]
q = all_quotes[sym]
print(f" {fmt_contract(q, sym, info)}")
print("\n" + "=" * 65)
# 关键比价分析
if "GC=F" in all_quotes and "CL=F" in all_quotes and "SI=F" in all_quotes:
au = all_quotes["GC=F"]
ag = all_quotes["SI=F"]
oil = all_quotes["CL=F"]
if ag["price"] > 0:
gsr = au["price"] / ag["price"] # 金银比
print(f"\n📐 关键比价:")
print(f" 金银比 (GC/SI): {gsr:.1f} (均值约80，高→银相对便宜)")
if oil["price"] > 0 and au["price"] > 0:
oil_au = oil["price"] / (au["price"] / 100) # 油价/金价(归一化)
print(f" 油金比: {oil_au:.2f} (高→通胀预期强，低→避险)")
# DXY 美元指数分析
dxy = all_quotes.get("DX-Y.NYB")
if dxy:
dxy_val = dxy["price"]
print(f"\n🌍 美元指数: {dxy_val:.2f} {dxy['pct']:+.2f}%")
if dxy_val > 104:
print(f" → 美元强势(>{104}) 大宗商品承压 黄金/原油倾向下跌 🟢")
elif dxy_val < 100:
print(f" → 美元弱势(<{100}) 大宗商品获支撑 黄金/原油倾向上涨 🔴")
else:
print(f" → 美元中性({100}~{104}) 大宗商品区间震荡")
if __name__ == "__main__":
print("正在获取期货数据...")
main()
全球汇率监控工具。当用户询问「美元汇率」「人民币汇率」「港币汇率」「日元汇率」「欧元汇率」「英镑汇率」「美元兑人民币」「USD/CNY」「汇率」「外汇」「离岸人民币」「在岸人民币」「换汇」时使用。使用 Yahoo Finance 获取实时汇率数据。
---
name: fx-radar
description: 全球汇率监控工具。当用户询问「美元汇率」「人民币汇率」「港币汇率」「日元汇率」「欧元汇率」「英镑汇率」「美元兑人民币」「USD/CNY」「汇率」「外汇」「离岸人民币」「在岸人民币」「换汇」时使用。使用 Yahoo Finance 获取实时汇率数据。
---
# 汇率雷达 (FX Radar)
## 覆盖品种
| 货币对 | Yahoo代码 | 说明 |
|--------|-----------|------|
| 美元/人民币 | USDCNY=X | USD/CNY |
| 美元/港元 | USDHKD=X | USD/HKD |
| 离岸人民币/港元 | CNHHKD=X | CNH/HKD |
| 美元/日元 | USDJPY=X | USD/JPY |
| 美元/瑞郎 | USDCHF=X | USD/CHF |
| 美元/加元 | USDCAD=X | USD/CAD |
| 欧元/美元 | EURUSD=X | EUR/USD |
| 英镑/美元 | GBPUSD=X | GBP/USD |
| 澳元/美元 | AUDUSD=X | AUD/USD |
| 纽元/美元 | NZDUSD=X | NZD/USD |
| 美元/离岸人民币 | USDCNH=X | USD/CNH |
## 数据获取
```python
import requests
FOREX_SYMBOLS = {
"USDCNY=X": "USD/CNY 美元/在岸人民币",
"USDHKD=X": "USD/HKD 美元/港元",
"USDJPY=X": "USD/JPY 美元/日元",
"EURUSD=X": "EUR/USD 欧元/美元",
"GBPUSD=X": "GBP/USD 英镑/美元",
"USDCHF=X": "USD/CHF 美元/瑞郎",
"AUDUSD=X": "AUD/USD 澳元/美元",
"NZDUSD=X": "NZD/USD 纽元/美元",
"USDCNH=X": "USD/CNH 美元/离岸人民币",
}
def get_forex():
results = {}
for sym, name in FOREX_SYMBOLS.items():
try:
url = f"https://query2.finance.yahoo.com/v8/finance/chart/{sym}?interval=1d&range=1d"
r = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=15)
meta = r.json()["chart"]["result"][0]["meta"]
price = meta["regularMarketPrice"]
prev = meta["chartPreviousClose"]
pct = (price - prev) / prev * 100
results[sym] = {"name": name, "price": price, "chg": price-prev, "pct": pct}
except Exception:
pass
return results
```
## 情报解读框架
| 货币对 | 关键阈值 | 信号 |
|--------|----------|------|
| USD/CNY | > 7.30 | 人民币贬值压力 |
| USD/CNY | < 7.10 | 人民币偏强 |
| USD/JPY | > 150 | 日本干预风险 |
| USD/JPY | < 145 | 日本央行容忍区间 |
| EUR/USD | > 1.10 | 欧元强势 |
| EUR/USD | < 1.05 | 欧元弱势 |
| USD/CNH vs USD/CNY | CNH大幅强于CNY | 资本管制收紧信号 |
**分析顺序：**
1. 美元指数（DXY）方向 — 全球货币锚
2. USD/CNY — 人民币核心价格
3. USD/HKD — 港币联系汇率稳定
4. USD/JPY — 日本央行政策信号
5. EUR/GBP/AUD — 发达市场货币对比
6. USD/CNH vs CNY利差 — 资本流动信号
## 快速查询命令
```bash
cd C:\Users\gold3\.openclaw\workspace\skills\fx-radar\scripts
# 查询全部汇率
python fx_spot.py
```
FILE:scripts/fx_spot.py
"""
汇率雷达 - 全球主要货币对实时行情
数据源: Yahoo Finance forex (USDCNY=X, USDHKD=X 等)
备用: European Central Bank (frankfurter.app)
"""
import requests
# Yahoo Finance 汇率代码
FOREX_PAIRS = {
# 亚洲
"USDCNY=X": ("USD/CNY", "💴 美元/在岸人民币", "🇨🇳 中国"),
"USDHKD=X": ("USD/HKD", "💴 美元/港元", "🇭🇰 香港"),
"USDCNH=X": ("USD/CNH", "💴 美元/离岸人民币", "🇭🇰 CNH"),
"USDJPY=X": ("USD/JPY", "💴 美元/日元", "🇯🇵 日本"),
"USDKRW=X": ("USD/KRW", "💴 美元/韩元", "🇰🇷 韩国"),
"USDSGD=X": ("USD/SGD", "💴 美元/新加坡元", "🇸🇬 新加坡"),
"USDTHB=X": ("USD/THB", "💴 美元/泰铢", "🇹🇭 泰国"),
"USDTWD=X": ("USD/TWD", "💴 美元/台币", "🇹🇼 台湾"),
"USDVND=X": ("USD/VND", "💴 美元/越南盾", "🇻🇳 越南"),
# 发达市场
"EURUSD=X": ("EUR/USD", "💶 欧元/美元", "🇪🇺 欧元区"),
"GBPUSD=X": ("GBP/USD", "💷 英镑/美元", "🇬🇧 英国"),
"AUDUSD=X": ("AUD/USD", "💳 澳元/美元", "🇦🇺 澳大利亚"),
"NZDUSD=X": ("NZD/USD", "🇳🇿 纽元/美元", "🇳🇿 新西兰"),
"USDCHF=X": ("USD/CHF", "💴 美元/瑞郎", "🇨🇭 瑞士"),
"USDCAD=X": ("USD/CAD", "💴 美元/加元", "🇨🇦 加拿大"),
# 黄金/商品货币
"DX-Y.NYB": ("DXY", "📊 美元指数(DXY)", "🌐 美元"),
}
# ECB 备用数据（frankfurter.app）
ECB_PAIRS = ["CNY", "HKD", "JPY", "EUR", "GBP", "AUD", "CHF", "CAD", "NZD", "KRW", "SGD", "THB"]
INVERTED_ECB_PAIRS = {"EUR", "GBP", "AUD", "NZD"}
def get_yahoo_forex():
"""Yahoo Finance 获取汇率（主数据源）"""
results = {}
import concurrent.futures
def _fetch_one(sym):
try:
url = f"https://query2.finance.yahoo.com/v8/finance/chart/{sym}?interval=1d&range=1d"
r = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=10)
r.raise_for_status()
d = r.json()
meta = d["chart"]["result"][0]["meta"]
price = meta.get("regularMarketPrice")
prev = meta.get("chartPreviousClose") or meta.get("previousClose")
if price is None or prev is None or prev == 0:
return None
chg = price - prev
pct = chg / prev * 100
name, label, flag = FOREX_PAIRS[sym]
return sym, {
"name": name, "label": label, "flag": flag,
"price": price, "chg": chg, "pct": pct
}
except Exception:
return None
with concurrent.futures.ThreadPoolExecutor(max_workers=6) as ex:
futures = {ex.submit(_fetch_one, sym): sym for sym in FOREX_PAIRS}
for f in concurrent.futures.as_completed(futures, timeout=20):
result = f.result()
if result is not None:
sym, data = result
results[sym] = data
return results
def get_ecb_forex():
"""欧洲央行汇率（备用数据源，不需要 key）"""
results = {}
try:
url = f"https://api.frankfurter.app/latest?base=USD&symbols={','.join(ECB_PAIRS)}"
r = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=15)
r.raise_for_status()
data = r.json().get("rates", {})
for curr, rate in data.items():
if curr in INVERTED_ECB_PAIRS:
if not rate:
continue
sym = f"{curr}USD=X"
name = f"{curr}/USD"
price = 1 / rate
else:
sym = f"USD{curr}=X"
name = f"USD/{curr}"
price = rate
results[sym] = {
"name": name, "label": name,
"flag": "🌐", "price": price,
"chg": 0, "pct": 0, "source": "ECB"
}
except (requests.RequestException, ValueError):
pass
return results
def fmt_row(item):
"""格式化一行汇率数据"""
label = item["label"]
price = item["price"]
pct = item.get("pct", 0)
if pct == 0:
pct_str = " --"
else:
pct_str = f"{pct:>+6.2f}%"
arrow = "🔴" if pct > 0 else "🟢" if pct < 0 else "⚪"
# 特殊处理：汇率小于1的显示更多小数位
if price and price < 0.01:
price_str = f"{price:.6f}"
elif price and price < 1:
price_str = f"{price:.4f}"
elif price and price < 100:
price_str = f"{price:.4f}"
else:
price_str = f"{price:.2f}"
return f" {arrow} {label:30s} {price_str:>12s} {pct_str}"
def main():
print("\n" + "=" * 70)
print(" 💱 汇率雷达 — 全球主要货币对")
print("=" * 70)
print("\n正在获取数据...")
yahoo_data = get_yahoo_forex()
ecb_data = get_ecb_forex()
# 合并：优先 Yahoo，备用 ECB
all_data = {**ecb_data, **yahoo_data}
if not all_data:
print("❌ 所有数据源均获取失败，请检查网络后重试")
return
# 分类显示
asia = ["USDCNY=X", "USDHKD=X", "USDCNH=X", "USDJPY=X", "USDKRW=X",
"USDSGD=X", "USDTHB=X", "USDTWD=X", "USDVND=X"]
majors = ["EURUSD=X", "GBPUSD=X", "AUDUSD=X", "NZDUSD=X", "USDCHF=X", "USDCAD=X"]
print("\n【 💴 亚洲货币 】")
for sym in asia:
if sym in all_data:
print(fmt_row(all_data[sym]))
print("\n【 💶 发达市场货币 】")
for sym in majors:
if sym in all_data:
print(fmt_row(all_data[sym]))
if "DX-Y.NYB" in all_data:
print("\n【 📊 美元指数 】")
print(fmt_row(all_data["DX-Y.NYB"]))
# 关键汇率解读
print("\n" + "=" * 70)
print("\n📐 关键汇率解读:")
signals = []
if "USDCNY=X" in all_data:
cny = all_data["USDCNY=X"]["price"]
if cny > 7.30:
signals.append(f"USD/CNY={cny:.4f} → 人民币偏弱，贬值压力 ⚠️")
elif cny < 7.10:
signals.append(f"USD/CNY={cny:.4f} → 人民币偏强 🟢")
else:
signals.append(f"USD/CNY={cny:.4f} → 人民币区间震荡")
if "USDJPY=X" in all_data:
jpy = all_data["USDJPY=X"]["price"]
if jpy > 150:
signals.append(f"USD/JPY={jpy:.2f} → 日本干预风险！⚠️")
elif jpy < 145:
signals.append(f"USD/JPY={jpy:.2f} → 日央行容忍区间")
else:
signals.append(f"USD/JPY={jpy:.2f} → 正常区间")
if "EURUSD=X" in all_data:
eur = all_data["EURUSD=X"]["price"]
if eur > 1.10:
signals.append(f"EUR/USD={eur:.4f} → 欧元强势 🔴")
elif eur < 1.05:
signals.append(f"EUR/USD={eur:.4f} → 欧元弱势 🟢")
else:
signals.append(f"EUR/USD={eur:.4f} → 欧元中性")
if "USDCNH=X" in all_data and "USDCNY=X" in all_data:
cnh = all_data["USDCNH=X"]["price"]
cny = all_data["USDCNY=X"]["price"]
diff = (cnh - cny) * 10000 # 转为pip
if diff > 50:
signals.append(f"CNH-CNY利差={diff:+.0f}pips → 离岸人民币弱于在岸，外流压力偏大 ⚠️")
elif diff < -50:
signals.append(f"CNH-CNY利差={diff:+.0f}pips → 离岸人民币强于在岸，离岸情绪偏稳 🟢")
for s in signals:
print(f" {s}")
if not signals:
print(" (数据不足，无法生成解读)")
print("\n" + "=" * 70)
if __name__ == "__main__":
main()
数字货币行情监控工具。当用户询问「比特币」「以太坊」「BTC」「ETH」「数字货币行情」「加密货币」「Crypto」「比特币价格」「以太坊价格」时使用。使用 Yahoo Finance（实时价格）和 CoinGecko（市值/交易量）双数据源。
---
name: crypto-radar
description: 数字货币行情监控工具。当用户询问「比特币」「以太坊」「BTC」「ETH」「数字货币行情」「加密货币」「Crypto」「比特币价格」「以太坊价格」时使用。使用 Yahoo Finance（实时价格）和 CoinGecko（市值/交易量）双数据源。
---
# 数字货币雷达 (Crypto Radar)
## 覆盖品种
| 币种 | 代码 | 类型 |
|------|------|------|
| 比特币 | BTC | 加密货币之王 |
| 以太坊 | ETH | 智能合约平台 |
| Solana | SOL | 高性能公链 |
| 币安币 | BNB | 交易所平台币 |
| XRP | XRP | Ripple 跨境支付 |
| 狗狗币 | DOGE | Meme 币 |
| 艾达币 | ADA | 卡尔达诺 |
## 数据获取
### Yahoo Finance（实时价格）
```python
import requests
def get_crypto_price(symbol):
url = f"https://query2.finance.yahoo.com/v8/finance/chart/{symbol}-USD?interval=1d&range=1d"
headers = {"User-Agent": "Mozilla/5.0"}
r = requests.get(url, headers=headers, timeout=10)
meta = r.json()["chart"]["result"][0]["meta"]
price = meta["regularMarketPrice"]
prev = meta["chartPreviousClose"]
chg = price - prev
pct = chg / prev * 100
return {"price": price, "chg": chg, "pct": pct}
```
### CoinGecko（市值 + 24h 交易量）
```python
import requests
def get_crypto_markets():
ids = "bitcoin,ethereum,solana,binancecoin,ripple,cardano,dogecoin"
url = f"https://api.coingecko.com/api/v3/coins/markets?vs_currency=usd&ids={ids}&order=market_cap_desc&per_page=10&sparkline=false&price_change_percentage=24h"
r = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=10)
return r.json()
# 返回: price, market_cap, total_volume, price_change_percentage_24h
```
## 情报解读框架
| 指标 | 信号 |
|------|------|
| BTC 市占率 > 55% | 避险情绪，主流资金抱团 |
| BTC 市占率 < 45% | Alt Season（山寨季）|
| ETH/BTC 比值 < 0.030 | ETH 相对 BTC 弱势 |
| ETH/BTC 比值 > 0.050 | ETH 强势，可能有 Alt Season |
| 山寨币涨幅 > BTC 3倍 | 资金外溢，风险偏好上升 |
| 全市场市值 vs BTC | 判断整体趋势方向 |
**关键比值：**
- ETH/BTC ratio：以太坊相对比特币强弱
- BTC Dominance：比特币市值占全市场比例
- 山寨季指数：全市场 vs BTC 的涨跌对比
**分析顺序：**
1. BTC 价格 + 市占率（方向）
2. ETH 价格 + ETH/BTC 比值（Alt 风向标）
3. 主流山寨（SOL/XRP/DOGE）— 情绪扩散
4. 全市场市值变化（判断趋势是否延续）
## 快速查询命令
```bash
cd C:\Users\gold3\.openclaw\workspace\skills\crypto-radar\scripts
# 查询全部加密货币行情
python crypto_spot.py
```
FILE:scripts/crypto_spot.py
"""
数字货币雷达 - 加密货币实时行情
双数据源：Yahoo Finance(价格) + CoinGecko(市值/交易量)
"""
import requests
# Yahoo Finance 实时价格
YAHOO_SYMBOLS = {
"BTC-USD": "比特币BTC",
"ETH-USD": "以太坊ETH",
"SOL-USD": "Solana",
"BNB-USD": "币安币BNB",
"XRP-USD": "XRP",
"DOGE-USD": "狗狗币DOGE",
"ADA-USD": "艾达币ADA",
"AVAX-USD": "雪崩AVAX",
"DOT-USD": "波卡DOT",
"MATIC-USD": "Polygon",
}
MACRO_SYMBOLS = {
"DX-Y.NYB": "DXY美元指数",
"^VIX": "VIX恐慌指数",
}
# CoinGecko ID 映射
COINGECKO_IDS = {
"BTC-USD": "bitcoin",
"ETH-USD": "ethereum",
"SOL-USD": "solana",
"BNB-USD": "binancecoin",
"XRP-USD": "ripple",
"DOGE-USD": "dogecoin",
"ADA-USD": "cardano",
"AVAX-USD": "avalanche-2",
"DOT-USD": "polkadot",
"MATIC-USD": "matic-network",
}
def get_yahoo_quote(sym):
url = f"https://query2.finance.yahoo.com/v8/finance/chart/{sym}?interval=1d&range=1d"
r = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=15)
r.raise_for_status()
meta = r.json()["chart"]["result"][0]["meta"]
price = meta.get("regularMarketPrice")
prev = meta.get("chartPreviousClose") or meta.get("previousClose")
if price is None or prev in (None, 0):
return None
chg = price - prev
pct = chg / prev * 100
return {"price": price, "chg": chg, "pct": pct}
def get_yahoo_quotes():
"""Yahoo Finance 获取实时价格（容错）"""
results = {}
for sym, name in YAHOO_SYMBOLS.items():
try:
quote = get_yahoo_quote(sym)
if quote:
results[sym] = {"name": name, **quote}
except (requests.RequestException, ValueError, KeyError, IndexError, TypeError):
continue
return results
def get_macro_quotes():
"""Yahoo Finance 获取宏观指标（DXY + VIX，容错）"""
results = {}
for sym, name in MACRO_SYMBOLS.items():
try:
quote = get_yahoo_quote(sym)
if quote:
results[sym] = {"name": name, **quote}
except (requests.RequestException, ValueError, KeyError, IndexError, TypeError):
continue
return results
def get_coingecko_markets():
"""CoinGecko 获取市值和交易量"""
ids = ",".join(COINGECKO_IDS.values())
url = (f"https://api.coingecko.com/api/v3/coins/markets"
f"?vs_currency=usd&ids={ids}&order=market_cap_desc"
f"&per_page=10&sparkline=false&price_change_percentage=24h")
try:
r = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=15)
if r.status_code == 200:
return {c["id"]: c for c in r.json()}
if r.status_code == 429:
return {}
except (requests.RequestException, ValueError, TypeError):
return {}
return {}
def fmt_row(name, price, pct, chg, mcap=None, vol=None, indent=""):
arrow = "🔴" if pct > 0 else "🟢" if pct < 0 else "⚪"
pct_str = f"{pct:>+7.2f}%"
price_str = f">12,.2f" if price >= 1 else f">12.6f"
chg_str = f"{chg:+.2f}" if price >= 1 else f"{chg:+.6f}"
line1 = f"{indent}{arrow} {name:12s} {price_str} {chg_str:>12s}"
if mcap is not None and vol is not None:
mc_str = f"MCap:>6.2fT" if mcap >= 1e12 else f"MCap:>6.2fB"
vol_str = f"Vol24h:>5.2fB"
line1 += f" {pct_str} {mc_str} {vol_str}"
else:
line1 += f" {pct_str}"
return line1
def get_coingecko_price_fallback(markets):
results = {}
for sym, coin_id in COINGECKO_IDS.items():
market = markets.get(coin_id)
if not market:
continue
price = market.get("current_price")
pct = market.get("price_change_percentage_24h")
chg = market.get("price_change_24h")
if price is None or pct is None or chg is None:
continue
results[sym] = {
"name": YAHOO_SYMBOLS[sym],
"price": price,
"chg": chg,
"pct": pct,
}
return results
def main():
print("\n" + "=" * 75)
print(" 🪙 数字货币雷达 — 实时行情")
print("=" * 75)
print("\n正在获取数据...")
quotes = get_yahoo_quotes()
markets = get_coingecko_markets()
fallback_quotes = get_coingecko_price_fallback(markets)
for sym, data in fallback_quotes.items():
quotes.setdefault(sym, data)
macro = get_macro_quotes()
# 当前跟踪币种市值汇总（非全市场）
total_mcap = sum(
(markets.get(cgid, {}).get("market_cap") or 0)
for cgid in COINGECKO_IDS.values()
)
btc_mcap = markets.get("bitcoin", {}).get("market_cap") or 0
eth_mcap = markets.get("ethereum", {}).get("market_cap") or 0
tracked_btc_share = btc_mcap / total_mcap * 100 if total_mcap > 0 else 0
# BTC/ETH/SOL 第一梯队
print("\n【 🥇 主流加密货币 】")
for sym in ["BTC-USD", "ETH-USD", "SOL-USD"]:
if sym not in quotes:
continue
q = quotes[sym]
cgid = COINGECKO_IDS.get(sym)
mkt = markets.get(cgid, {}) if cgid else {}
print(fmt_row(q["name"], q["price"], q["pct"], q["chg"],
mkt.get("market_cap"), mkt.get("total_volume")))
# 平台币 + 山寨
tier2 = ["BNB-USD", "XRP-USD", "ADA-USD", "DOGE-USD"]
tier2_names = {s: quotes.get(s) for s in tier2}
if any(tier2_names.values()):
print("\n【 🥈 平台币 & 山寨 】")
for sym in tier2:
if sym not in quotes:
continue
q = quotes[sym]
cgid = COINGECKO_IDS.get(sym)
mkt = markets.get(cgid, {}) if cgid else {}
print(fmt_row(q["name"], q["price"], q["pct"], q["chg"],
mkt.get("market_cap"), mkt.get("total_volume")))
# L2 / 其他
tier3 = ["AVAX-USD", "DOT-USD", "MATIC-USD"]
tier3_data = {s: quotes.get(s) for s in tier3}
if any(tier3_data.values()):
print("\n【 🥉 L2 & 波卡生态 】")
for sym in tier3:
if sym not in quotes:
continue
q = quotes[sym]
cgid = COINGECKO_IDS.get(sym)
mkt = markets.get(cgid, {}) if cgid else {}
print(fmt_row(q["name"], q["price"], q["pct"], q["chg"],
mkt.get("market_cap"), mkt.get("total_volume")))
# 关键比值分析
print("\n" + "=" * 75)
print("\n📐 关键比值:")
print(f" 跟踪篮子中 BTC 市值占比: {tracked_btc_share:.1f}%")
if "ETH-USD" in quotes and "BTC-USD" in quotes:
eth_btc = quotes["ETH-USD"]["price"] / quotes["BTC-USD"]["price"]
print(f" ETH/BTC: {eth_btc:.5f} (<0.030弱势 / >0.050强势)")
print(f" 跟踪篮子总市值: .3fT")
# 宏观数据
if macro:
print("\n🌍 宏观背景:")
for sym, m in macro.items():
arrow = "🔴" if m["pct"] > 0 else "🟢" if m["pct"] < 0 else "⚪"
print(f" {arrow} {m['name']}: {m['price']:.2f} {m['pct']:+.2f}%")
# DXY 信号解读
dxy = macro.get("DX-Y.NYB")
if dxy:
dxy_val = dxy["price"]
if dxy_val > 104:
print(f" → DXY强势(>{104}) 美元强 → 加密货币承压 🟢")
elif dxy_val < 100:
print(f" → DXY弱势(<{100}) 美元弱 → 加密货币利好 🔴")
else:
print(f" → DXY中性({100}~{104}) 加密货币区间震荡")
vix = macro.get("^VIX")
if vix:
vix_val = vix["price"]
if vix_val > 25:
print(f" → VIX高({vix_val:.0f}) 市场恐慌 → 避险优先 🟢BTC")
elif vix_val < 15:
print(f" → VIX低({vix_val:.0f}) 风险偏好 → 山寨有机会 🔴Altcoin")
else:
print(f" → VIX中性({vix_val:.0f}) 平衡市")
print("\n" + "=" * 75)
if __name__ == "__main__":
main()
港股行情与舆情监控工具。当用户询问「港股怎么样」「恒生指数」「港股大盘」「港股涨跌」「HK股」「港股行情监控」「南向资金」时使用。支持东方财富港股API、新浪财经港股接口、Yahoo Finance获取实时行情，以及Google News RSS和X/Twitter舆情监控。
---
name: hk-stock-radar
description: 港股行情与舆情监控工具。当用户询问「港股怎么样」「恒生指数」「港股大盘」「港股涨跌」「HK股」「港股行情监控」「南向资金」时使用。支持东方财富港股API、新浪财经港股接口、Yahoo Finance获取实时行情，以及Google News RSS和X/Twitter舆情监控。
---
# 港股雷达 (HK-Stock Radar)
## 数据源总览
| 数据源 | 用途 | 稳定性 |
|--------|------|--------|
| 东方财富港股板块API | 行业/概念板块涨跌排行、热点追踪 | ⭐⭐⭐ |
| 新浪财经港股接口 | 个股实时行情（最稳定） | ⭐⭐⭐ |
| Yahoo Finance | 恒生指数实时行情 | ⭐⭐ |
| akshare | 港股实时行情、沪深港通 | ⭐⭐ |
## 实时行情查询
### 恒生指数实时行情
```python
import requests
def get_hsi():
"""恒生指数 + 国企指数 + 恒生科技"""
url = "https://hq.sinajs.cn/list=hkHSI,hkHSTECH,hkHSCEI"
headers = {"Referer": "http://finance.sina.com.cn"}
r = requests.get(url, headers=headers, timeout=10)
r.encoding = 'gbk'
results = {}
for line in r.text.strip().split('\n'):
if '=' not in line:
continue
_, data = line.split('=')
vals = data.replace('"', '').replace(';', '').split(',')
if len(vals) < 6:
continue
name = vals[0]
price = float(vals[1])
chg = float(vals[4])
pct = float(vals[5])
results[name] = {"price": price, "chg": chg, "pct": pct}
return results
```
### 新浪财经港股个股行情
```python
import requests
def get_hk_quote(codes):
"""查询港股实时行情
codes: str or list, e.g. 'hk00700' or ['hk00700', 'hk09988']
"""
if isinstance(codes, str):
codes = [codes]
url = f"https://hq.sinajs.cn/list={','.join(codes)}"
headers = {"Referer": "http://finance.sina.com.cn"}
r = requests.get(url, headers=headers, timeout=10)
r.encoding = 'gbk'
results = []
for line in r.text.strip().split('\n'):
if '=' not in line:
continue
_, data = line.split('=')
code = _[-7:].replace('"', '') # e.g. hk00700
vals = data.replace('"', '').replace(';', '').split(',')
if len(vals) < 6:
continue
try:
name = vals[0]
price = float(vals[1])
prev = float(vals[2])
chg = price - prev
pct = chg / prev * 100
high = float(vals[4])
low = float(vals[5])
volume = float(vals[3]) / 1e6 # 成交量（手）
arrow = "🔴" if pct > 0 else "🟢" if pct < 0 else "⚪"
results.append({
"code": code, "name": name, "price": price,
"chg": chg, "pct": pct, "high": high, "low": low,
"volume": volume, "arrow": arrow
})
except (ValueError, IndexError):
continue
return results
```
### 东方财富港股板块 API
```python
import requests
def get_hk_sector_ranking():
"""港股行业板块涨跌排行"""
url = "http://push2.eastmoney.com/api/qt/clist/get"
params = {
"pn": 1, "pz": 30, "po": 1, "np": 1,
"fltt": 2, "invt": 2,
"fid": "f3",
"fs": "m:1+t:23", # 港股行业板块
"fields": "f12,f14,f2,f3,f5,f6"
}
headers = {"Referer": "http://quote.eastmoney.com/"}
r = requests.get(url, params=params, headers=headers, timeout=10)
diff = r.json()["data"]["diff"]
return [{"板块": x["f14"], "现价": x["f2"], "涨跌幅": x["f3"],
"成交额": x["f6"]} for x in diff]
```
### 沪深港通（南向资金）
```python
import akshare as ak
def get_southbound_flow():
"""南向资金净流入"""
df = ak.stock_hsgt_north_net_flow_em()
# 沪深港通北向资金
return df.tail(5) # 最近5个交易日
```
## 主流港股代码速查
| 股票 | 代码 | 名称 |
|------|------|------|
| 腾讯 | hk00700 | 腾讯控股 |
| 阿里 | hk09988 | 阿里巴巴 |
| 美团 | hk03690 | 美团 |
| 比亚迪 | hk01211 | 比亚迪股份 |
| 京东 | hk09618 | 京东集团 |
| 小米 | hk01810 | 小米集团 |
| 恒生指数 | hkHSI | 恒生指数 |
| 恒生科技 | hkHSTECH | 恒生科技指数 |
| 国企指数 | hkHSCEI | 恒生国企指数 |
## 舆情监控
### Google News Live（突发新闻）
```
https://news.google.com/rss/search?q=港股+恒生+今日+when:1h
https://news.google.com/rss/search?q=香港股市+2026+when:1h
https://news.google.com/rss/search?q=南向资金+港股+when:1h
```
### X/Twitter 港股舆情
使用 browser 工具访问已登录的 @bearfrom2077：
```
https://x.com/search?q=港股%20恒生指数&f=live
https://x.com/search?q=南向资金&f=live
https://x.com/search?q=hkstocks%20hangseng&f=live
```
**核心关键词组合：**
- `港股 恒生` — 大盘情绪
- `南向资金` — 外资态度
- `HK IPO` — 新股动态
- `科技股 港股` — 板块热点
## 情报解读框架
| 指标 | 阈值 | 信号 |
|------|------|------|
| 恒生指数跌幅 | > 1.5% | 系统性风险预警 |
| 南向资金净流入 | > 50亿/日 | 内地资金抄底 |
| 南向资金净流出 | > 30亿/日 | 谨慎信号 |
| 腾讯/阿里/美团 | 同时下跌 > 2% | 科技股出逃 |
| 防御板块（银行/公用）领涨 | 资金抱团 | 非系统性风险 |
**分析顺序：**
1. 恒生指数 + 国企指数 + 恒生科技（大盘方向）
2. 科技股（腾讯/阿里/美团/小米）— 港股主线
3. 南向资金（内地钱往哪走）
4. 板块涨跌（资金在哪里）
5. 交叉验证 + 给出判断
## Cron 配置建议
| 频率 | 内容 | 适用场景 |
|------|------|----------|
| 每15分钟 | 恒生指数 + 恒生科技 | 盘中监控 |
| 每30分钟 | 科技股四巨头（腾讯/阿里/美团/小米） | 港股主线 |
| 每小时 | 港股板块排行 + 舆情 | 热点追踪 |
| 有问才查 | 个股行情 | 被动触发 |
## 快速查询命令
```bash
cd C:\Users\gold3\.openclaw\workspace\skills\hk-stock-radar\scripts
# 恒生指数
python hk_index.py
# 个股行情（传入港股代码）
python hk_quote.py hk00700
# 港股板块
python hk_sector.py
# 南向资金
python southbound.py
```
FILE:scripts/dashboard.py
"""
港股雷达仪表盘：一键汇总恒生指数 + 科技股 + 板块
用法: python dashboard.py
"""
import subprocess
import sys
import os
base = os.path.dirname(os.path.abspath(__file__))
print("=" * 50)
print(" 🦞 港股雷达仪表盘")
print("=" * 50)
print("\n>>> 恒生指数")
subprocess.run([sys.executable, os.path.join(base, "hk_index.py")])
print("\n>>> 港股科技四巨头 (腾讯/阿里/美团/小米)")
subprocess.run([sys.executable, os.path.join(base, "hk_quote.py"),
"hk00700", "hk09988", "hk03690", "hk01810"])
print("\n>>> 港股板块涨跌TOP10")
subprocess.run([sys.executable, os.path.join(base, "hk_sector.py"), "30"])
FILE:scripts/hk_index.py
"""
港股主要指数实时行情
恒生指数、恒生科技、恒生国企指数
"""
import requests
HK_INDICES = {
"hkHSI": "恒生指数",
"hkHSTECH": "恒生科技",
"hkHSCEI": "恒生国企指数",
}
def get_hk_indices():
codes = ",".join(HK_INDICES.keys())
url = f"https://hq.sinajs.cn/list={codes}"
headers = {"Referer": "http://finance.sina.com.cn"}
try:
r = requests.get(url, headers=headers, timeout=10)
r.raise_for_status()
except requests.RequestException:
return {}
r.encoding = 'gbk'
results = {}
for line in r.text.strip().split('\n'):
if '=' not in line:
continue
key, data = line.split('=', 1)
code = key.split('_')[-1].replace('"', '') # e.g. hkHSI
if code not in HK_INDICES:
continue
vals = data.replace('"', '').replace(';', '').split(',')
if len(vals) < 9:
continue
try:
name = vals[1] # 恒生指数
price = float(vals[6])
prev = float(vals[3])
if prev == 0:
continue
chg = price - prev
pct = chg / prev * 100
arrow = "🔴" if pct > 0 else "🟢" if pct < 0 else "⚪"
results[code] = f"{arrow} {name}: {price:.2f} {chg:+.2f}({pct:+.2f}%)"
except (ValueError, IndexError):
continue
return results
if __name__ == "__main__":
print("\n=== 港股主要指数 ===")
indices = get_hk_indices()
for v in indices.values():
print(v)
FILE:scripts/hk_quote.py
"""
新浪财经港股个股实时行情
用法: python hk_quote.py hk00700
python hk_quote.py hk00700 hk09988 hk03690
"""
import requests
import sys
def get_hk_quote(codes):
if isinstance(codes, str):
codes = [codes]
url = f"https://hq.sinajs.cn/list={','.join(codes)}"
headers = {"Referer": "http://finance.sina.com.cn"}
r = requests.get(url, headers=headers, timeout=10)
r.encoding = 'gbk'
for line in r.text.strip().split('\n'):
if '=' not in line:
continue
key, data = line.split('=')
code = key.split('_')[-1].replace('"', '') # e.g. hk00700
vals = data.replace('"', '').replace(';', '').split(',')
if len(vals) < 6:
continue
try:
# 新浪港股格式: 0=代码,1=名称,2=现价,3=昨收,4=今开,5=高,6=低,7=涨跌额,8=涨跌幅%
name = vals[1]
price = float(vals[2])
prev = float(vals[3])
chg = price - prev
pct = chg / prev * 100
high = float(vals[5])
low = float(vals[6])
vol = float(vals[12]) / 1e4 if len(vals) > 12 else 0 # 成交量（手）
arrow = "🔴" if pct > 0 else "🟢" if pct < 0 else "⚪"
print(f"{arrow} {name:12s} {code:8s} 现价:{price:.2f} "
f"涨跌:{chg:+.2f}({pct:+.2f}%) 高:{high:.2f} 低:{low:.2f} 量:{vol:.0f}手")
except (ValueError, IndexError):
continue
if __name__ == "__main__":
if len(sys.argv) < 2:
print("用法: python hk_quote.py hk00700 [hk09988 ...]")
sys.exit(1)
get_hk_quote(sys.argv[1:])
FILE:scripts/hk_sector.py
"""
东方财富港股行业板块涨跌榜
"""
import requests
import sys
def get_hk_sector_ranking(top=20):
url = "http://push2.eastmoney.com/api/qt/clist/get"
params = {
"pn": 1, "pz": top, "po": 1, "np": 1,
"fltt": 2, "invt": 2,
"fid": "f3",
"fs": "m:1+t:23", # 港股行业板块
"fields": "f12,f14,f2,f3,f5,f6,f8"
}
headers = {"Referer": "http://quote.eastmoney.com/"}
r = requests.get(url, params=params, headers=headers, timeout=10)
r.raise_for_status()
diff = r.json()["data"]["diff"]
return diff
def format_sector(s):
pct = s["f3"]
arrow = "🔴" if pct > 0 else "🟢" if pct < 0 else "⚪"
return f"{arrow} {s['f14']:15s} {'+' if pct>=0 else ''}{pct}% 成交额:{s['f6']/1e8:.1f}亿"
if __name__ == "__main__":
top = int(sys.argv[1]) if len(sys.argv) > 1 else 20
print("\n=== 港股板块涨跌榜 ===")
data = get_hk_sector_ranking(top)
print("\n【涨幅榜】")
for s in sorted(data, key=lambda x: x["f3"], reverse=True)[:10]:
print(format_sector(s))
print("\n【跌幅榜】")
for s in sorted(data, key=lambda x: x["f3"])[:10]:
print(format_sector(s))
FILE:scripts/southbound.py
"""
南向资金（沪深港通港股通）净流入
需要 akshare: pip install akshare
"""
try:
import akshare as ak
import pandas as pd
except ImportError as e:
print(f"需要安装akshare: pip install akshare ({e})")
exit(1)
def get_southbound():
print("\n=== 南向资金（港股通）===")
try:
# 沪深港通南向资金历史
df = ak.stock_hsgt_north_net_flow_em()
print(df.tail(5).to_string(index=False))
except Exception as e:
print(f"南向资金获取失败: {e}")
print("\n尝试替代方案...")
try:
# 替代：东方财富南向资金
url = "https://push2.eastmoney.com/api/qt/kamt.rtmin/get"
params = {"fields1": "f1,f2,f3,f4", "fields2": "f51,f52,f53,f54,f55,f56"}
r = requests.get(url, params=params, timeout=10)
print(r.json())
except Exception as e2:
print(f"替代方案也失败: {e2}")
if __name__ == "__main__":
get_southbound()
A股综合监控与分析工具。当用户询问「A股今天怎么样」「大盘如何」「哪些板块在动」「今日涨跌」「北向资金」「涨停/跌停」「市场情绪」「龙虎榜」「现在适合上仓位吗」「分析A股」「A股复盘」「当前主线是什么」「A股核心企业」「茅台/宁德时代/比亚迪」「SHIBOR」「LPR」「中美10Y利差」「社融/人民币贷款」「A股...
---
name: a-stock-radar
description: A股综合监控与分析工具。当用户询问「A股今天怎么样」「大盘如何」「哪些板块在动」「今日涨跌」「北向资金」「涨停/跌停」「市场情绪」「龙虎榜」「现在适合上仓位吗」「分析A股」「A股复盘」「当前主线是什么」「A股核心企业」「茅台/宁德时代/比亚迪」「SHIBOR」「LPR」「中美10Y利差」「社融/人民币贷款」「A股宏观数据」时使用。覆盖实时行情监控、宏观快照、核心企业篮子、龙虎榜数据、短线情绪判断、深度量化分析。
---
# A股雷达 (A-Stock Radar)
`a-stock-radar` 现在是统一的 A 股入口，内部按四层能力组织：
1. `实时行情监控`
2. `宏观快照与核心企业`
3. `龙虎榜数据`
4. `短线情绪判断`
5. `深度量化分析`
## 1. 实时行情监控
适用问题：
- A股今天怎么样
- 哪些板块在动
- 北向资金如何
- 涨停跌停情况
- 某只股票现在多少钱
主要脚本：
- `scripts/index_spot.py`：主要指数
- `scripts/sector_ranking.py`：板块排行
- `scripts/stock_quote.py`：个股行情
- `scripts/zt_pool.py`：涨停池/跌停池
- `scripts/dashboard.py`：综合看盘入口
- `scripts/macro_snapshot.py`：SHIBOR、LPR、中美10Y利差、北向、新增人民币贷款
- `scripts/core_companies.py`：A股核心企业篮子
主要数据源：
- 东方财富：板块/排行
- 新浪财经：指数与个股实时行情
- akshare：涨跌停池、北向等扩展数据
## 2. 宏观快照与核心企业
适用问题：
- A股宏观数据怎么看
- SHIBOR/LPR 现在是什么水平
- 中美10Y利差如何
- 北向资金和融资需求怎么样
- 茅台/宁德时代/比亚迪现在怎么样
- 给我看A股核心企业篮子
主要脚本：
- `scripts/macro_snapshot.py`
- `scripts/core_companies.py`
覆盖维度：
- SHIBOR（银行间流动性）
- LPR（贷款市场报价利率）
- 中美10Y国债利差
- 北向资金净流入
- 新增人民币贷款
- 贵州茅台、宁德时代、比亚迪、招商银行、工商银行、美的集团、中芯国际、立讯精密
## 3. 龙虎榜数据
适用问题：
- 今日龙虎榜有哪些
- 机构净买入哪些股票
- 龙虎榜机构和散户买卖方向
- 哪些股票被机构大额卖出
主要脚本：
- `scripts/lhb_list.py`
数据来源：东方财富数据中心（akshare 接口）
输出内容：
- 机构净买入 Top10（含收盘价、涨跌幅、净买额、换手率、上榜原因）
- 机构净卖出 Top10（含收盘价、涨跌幅、净卖额、换手率、上榜原因）
- 数据日期标注（区分上榜日和数据更新日）
## 4. 短线情绪判断
适用问题：
- 今天市场情绪怎么样
- 现在是冰点还是亢奋
- 适不适合上仓位
- 短线环境强不强
主要脚本：
- `scripts/sentiment_snapshot.py`
核心指标：
- 涨停家数
- 跌停家数
- 炸板率
- 连板高度
输出内容：
- 情绪阶段：`冰点 / 分歧 / 修复 / 亢奋`
- 仓位建议：`空仓 / 轻仓 / 半仓 / 重仓`
- 短线打法建议
## 5. 深度量化分析
适用问题：
- 分析A股
- 做一份A股复盘
- 当前主线是什么
- 用量化视角看A股
主要脚本：
- `scripts/quant_analysis.py`
分析框架：
- 宏观流动性
- 政策主线
- ETF/龙虎榜/换手结构
- 情绪与筹码共振
输出内容：
- 主线判断
- 择时建议
- 复盘框架
- 风险提示
## 使用原则
- 只问盘面与行情：优先走 `实时行情监控`
- 问仓位与短线环境：优先走 `短线情绪判断`
- 问中高层判断与复盘：优先走 `深度量化分析`
## 迁移说明
以下旧 skill 已并入本 skill：
- `a-stock-market-sentiment`
- `ashare-quant`
FILE:scripts/browser_fetch.py
"""
浏览器备用抓取工具
当 API 请求失败时，通过抓取网页 HTML 解析数据。
纯 requests + BeautifulSoup，不依赖 Selenium。
"""
import requests
import time
import io
import contextlib
from typing import Optional, Callable, Any
class FetchError(Exception):
"""抓取失败异常"""
pass
def fetch_with_retry(
url: str,
headers: Optional[dict] = None,
timeout: int = 10,
retries: int = 3,
backoff: float = 1.5,
encoding: Optional[str] = None,
parser: Optional[str] = "html.parser",
session: Optional[requests.Session] = None,
) -> Any:
"""
带重试的抓取，同时返回 (text, soup)。
Args:
url: 目标URL
headers: 请求头
timeout: 单次超时（秒）
retries: 最大重试次数
backoff: 退避倍数
encoding: 响应编码（默认自动检测）
parser: BeautifulSoup 解析器
session: requests.Session（可复用连接）
Returns:
(response_text, BeautifulSoup)
Raises:
FetchError: 所有重试失败后
"""
default_headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
}
if headers:
default_headers.update(headers)
if session is None:
session = requests.Session()
last_error = None
for attempt in range(retries):
try:
r = session.get(url, headers=default_headers, timeout=timeout)
r.raise_for_status()
text = r.text
if encoding:
text = text.encode(r.encoding or "utf-8").decode(encoding, errors="replace")
try:
from bs4 import BeautifulSoup
soup = BeautifulSoup(text, parser)
except ImportError:
soup = None
return text, soup
except Exception as e:
last_error = e
if attempt < retries - 1:
wait = backoff ** attempt
time.sleep(wait)
continue
raise FetchError(f"全部{retries}次重试失败: {last_error}")
def parse_eastmoney_table(url: str, retries: int = 2) -> list:
"""
抓取东方财富网页表格数据。
Args:
url: 东方财富页面URL
retries: 重试次数
Returns:
list[dict]，每行一个字典
"""
try:
from bs4 import BeautifulSoup
except ImportError:
raise FetchError("需要安装 bs4: pip install beautifulsoup4")
text, soup = fetch_with_retry(url, retries=retries, parser="lxml")
tables = soup.find_all("table")
results = []
for table in tables:
try:
import pandas as pd
dfs = pd.read_html(io.StringIO(str(table)))
if dfs:
results.extend(dfs)
except Exception:
continue
return results
# ─── 常用东方财富行情页面 ────────────────────────────────────────────
INDEX_PAGE = "https://quote.eastmoney.com/center/gridlist.html"
def fetch_eastmoney_index() -> dict:
"""
备用：从东方财富网页抓取A股主要指数。
当 Sina hq API 不可用时调用。
Returns: dict {代码: {"name": str, "price": float, "chg": float, "pct": float}}
"""
from bs4 import BeautifulSoup
# 东方财富 A股指数页面
url = "https://push2.eastmoney.com/api/qt/clist/get?pn=1&pz=20&po=1&np=1&fltt=2&invt=2&fid=f3&fs=m:1+t:2&fields=f2,f3,f4,f12,f14"
headers = {"Referer": "https://quote.eastmoney.com/", "User-Agent": "Mozilla/5.0"}
try:
r = requests.get(url, headers=headers, timeout=10)
r.raise_for_status()
data = r.json()
items = data.get("data", {}).get("diff", [])
result = {}
name_map = {
"000001": "上证指数", "399001": "深证成指", "399006": "创业板指",
"000688": "科创50", "000300": "沪深300", "399905": "中证500",
}
for item in items:
code = str(item.get("f12", ""))
if code in name_map:
result[code] = {
"name": name_map[code],
"price": item.get("f2", 0),
"pct": item.get("f3", 0),
}
return result
except Exception:
pass
# 兜底：直接解析 Eastmoney 行情页 HTML
try:
text, soup = fetch_with_retry(
"https://quote.eastmoney.com/center/gridlist.html",
headers={"Referer": "https://www.eastmoney.com/"},
retries=2,
)
# 解析指数名称和价格
import re
indices = {}
# 东方财富页面里含实时数据在 JS 变量中
script_texts = soup.find_all("script")
for script in script_texts:
content = script.string or ""
# 找 hq_str_xxx 格式的数据
matches = re.findall(r'hq_str_(sh\d{6}|sz\d{6})="([^"]+)"', content)
for code, data in matches:
parts = data.split(",")
if len(parts) > 4:
try:
price = float(parts[3])
prev = float(parts[2])
pct = (price - prev) / prev * 100 if prev else 0
indices[code] = {"price": price, "pct": pct}
except ValueError:
continue
return indices
except Exception:
return {}
if __name__ == "__main__":
# 测试
print("测试 browser_fetch...")
result = fetch_eastmoney_index()
print(f"获取到 {len(result)} 个指数")
for v in result.values():
print(v)
FILE:scripts/core_companies.py
"""
A股核心企业篮子
贵州茅台 / 宁德时代 / 比亚迪 / 招商银行 / 工商银行 / 美的集团 / 中芯国际 / 立讯精密
"""
import os
import sys
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
if SCRIPT_DIR not in sys.path:
sys.path.insert(0, SCRIPT_DIR)
from stock_quote import fetch_stock_quotes, format_stock_quote
CORE_COMPANIES = {
"sh600519": "贵州茅台",
"sz300750": "宁德时代",
"sz002594": "比亚迪",
"sh600036": "招商银行",
"sh601398": "工商银行",
"sz000333": "美的集团",
"sh688981": "中芯国际",
"sz002475": "立讯精密",
}
if __name__ == "__main__":
print("\n=== 🏭 A股核心企业 ===")
quotes = fetch_stock_quotes(list(CORE_COMPANIES.keys()))
for item in quotes:
print(format_stock_quote(item))
FILE:scripts/dashboard.py
"""
A股雷达仪表盘：一键汇总主要指数 + 宏观快照 + 核心企业 + 板块涨跌
用法: python dashboard.py
"""
import subprocess
import sys
import os
base = os.path.dirname(os.path.abspath(__file__))
print("=" * 50)
print(" 🦞 A股雷达仪表盘")
print("=" * 50)
# 1. 主要指数
print("\n>>> 主要指数")
subprocess.run([sys.executable, os.path.join(base, "index_spot.py")])
# 2. 宏观快照
print("\n>>> 宏观快照")
subprocess.run([sys.executable, os.path.join(base, "macro_snapshot.py")])
# 3. 核心企业
print("\n>>> A股核心企业")
subprocess.run([sys.executable, os.path.join(base, "core_companies.py")])
# 4. 行业板块
print("\n>>> 行业涨幅榜TOP10")
result = subprocess.run([sys.executable, os.path.join(base, "sector_ranking.py"), "行业板块"], capture_output=True, text=True)
if result.returncode != 0 or "暂时不可用" in result.stdout or "失败" in result.stdout or not result.stdout.strip():
print("（行业板块数据暂时不可用，请稍后重试）")
else:
print(result.stdout, end="")
# 5. 龙虎榜
print("\n>>> 龙虎榜")
subprocess.run([sys.executable, os.path.join(base, "lhb_list.py")])
# 6. 短线情绪
print("\n>>> 短线情绪")
result2 = subprocess.run([sys.executable, os.path.join(base, "sentiment_snapshot.py")], capture_output=True, text=True)
print(result2.stdout, end="")
FILE:scripts/index_spot.py
"""
A股主要指数实时行情
上证、深成、创业板、科创50、沪深300
【双层架构】
1. 主：新浪财经 hq.sinajs.cn API
2. 备：东方财富网页 + 浏览器抓取（API 失败时自动切换）
"""
import requests
import time
MAJOR_INDICES = {
"sh000001": "上证指数",
"sz399001": "深证成指",
"sz399006": "创业板指",
"sh000688": "科创50",
"sh000300": "沪深300",
"sz399905": "中证500",
}
_NAME_MAP = {v: k for k, v in MAJOR_INDICES.items()}
def _get_via_sina() -> dict:
"""新浪财经 API（主）"""
codes = ",".join(MAJOR_INDICES.keys())
url = f"https://hq.sinajs.cn/list={codes}"
headers = {"Referer": "http://finance.sina.com.cn"}
r = requests.get(url, headers=headers, timeout=10)
r.raise_for_status()
r.encoding = 'gbk'
results = {}
for line in r.text.strip().split('\n'):
if '=' not in line:
continue
key, data = line.split('=', 1)
vals = data.replace('"', '').replace(';', '').split(',')
if len(vals) < 4:
continue
try:
code = key[-8:].replace('_hq_str', '').replace('"', '')
prev = float(vals[2])
price = float(vals[3])
if prev == 0:
continue
pct = (price - prev) / prev * 100
chg = price - prev
arrow = "🔴" if pct > 0 else "🟢" if pct < 0 else "⚪"
name = MAJOR_INDICES.get(code, code)
results[code] = f"{arrow} {name}: {price:.2f} {chg:+.2f}({pct:+.2f}%)"
except (ValueError, IndexError):
continue
return results
def _get_via_eastmoney() -> dict:
"""东方财富 API（备用1）"""
url = (
"https://push2.eastmoney.com/api/qt/clist/get"
"?pn=1&pz=20&po=1&np=1&fltt=2&invt=2&fid=f3"
"&fs=m:1+t:2&fields=f2,f3,f4,f12,f14"
)
headers = {
"Referer": "https://quote.eastmoney.com/",
"User-Agent": "Mozilla/5.0",
}
r = requests.get(url, headers=headers, timeout=10)
r.raise_for_status()
data = r.json()
items = data.get("data", {}).get("diff", [])
results = {}
for item in items:
code = "sh" + str(item.get("f12", ""))
name = MAJOR_INDICES.get(code)
if not name:
continue
price = item.get("f2", 0)
pct = item.get("f3", 0)
chg = price * pct / 100
arrow = "🔴" if pct > 0 else "🟢" if pct < 0 else "⚪"
results[code] = f"{arrow} {name}: {price:.2f} {chg:+.2f}({pct:+.2f}%)"
return results
def get_major_indices() -> dict:
"""
获取A股主要指数，尝试顺序：
新浪API → 东方财富API → 返回空dict（不抛异常）
"""
# 尝试新浪（主）
try:
result = _get_via_sina()
if result:
return result
except Exception:
pass
# 等待1秒再试东方财富（避免同时失败）
time.sleep(1)
# 尝试东方财富（备）
try:
result = _get_via_eastmoney()
if result:
return result
except Exception:
pass
return {}
if __name__ == "__main__":
print("\n=== A股主要指数 ===")
indices = get_major_indices()
for v in indices.values():
print(v)
FILE:scripts/lhb_list.py
"""
龙虎榜数据 - 东方财富数据中心
akshare: stock_lhb_detail_em(start_date, end_date)
"""
import akshare as ak
import pandas as pd
import sys
from datetime import datetime, timedelta
def format_amount(val):
"""格式化金额（亿元）"""
if abs(val) >= 1e8:
return f"{val/1e8:.2f}亿"
elif abs(val) >= 1e4:
return f"{val/1e4:.2f}万"
return f"{val:.0f}"
def get_lhb_list(days=3, top=15):
"""获取最近N日的龙虎榜数据"""
end_date = datetime.now().strftime("%Y%m%d")
start_date = (datetime.now() - timedelta(days=days)).strftime("%Y%m%d")
df = ak.stock_lhb_detail_em(start_date=start_date, end_date=end_date)
# 取最新上榜日的数据
latest_date = df['上榜日'].max()
df_latest = df[df['上榜日'] == latest_date].copy()
# 去重：同一股票同一日只留一条（取净买额最大的）
df_latest = df_latest.sort_values('龙虎榜净买额', key=abs, ascending=False).drop_duplicates(subset=['代码', '上榜日'])
# 分离净买入和净卖出
df_buy = df_latest[df_latest['龙虎榜净买额'] > 0].nlargest(top, '龙虎榜净买额')
df_sell = df_latest[df_latest['龙虎榜净买额'] < 0].nsmallest(top, '龙虎榜净买额')
return df_buy, df_sell, latest_date
def print_lhb():
print(f"\n=== 🐉 龙虎榜（最近3日, 上榜日: {datetime.now().strftime('%Y-%m-%d')}）===\n")
try:
df_buy, df_sell, latest_date = get_lhb_list(days=3, top=10)
except Exception as e:
print(f"获取龙虎榜数据失败: {e}")
return
print(f"📅 数据日期: {latest_date}\n")
# 净买入榜
print("【机构净买入 Top10】")
print(f"{'代码':<8} {'名称':<10} {'收盘价':>8} {'涨跌幅':>8} {'净买额':>10} {'换手率':>7} {'上榜原因'}")
print("-" * 90)
for _, row in df_buy.iterrows():
code = row['代码']
name = row['名称']
close = row['收盘价']
chg = row['涨跌幅']
net = row['龙虎榜净买额']
turnover = row['换手率']
reason = row['上榜原因'][:20] if pd.notna(row['上榜原因']) else ''
arrow = '🔴' if chg > 0 else '🟢'
print(f"{arrow}{code:<6} {name:<10} {close:>8.2f} {chg:>+7.2f}% {format_amount(net):>10} {turnover:>6.1f}% {reason}")
print()
print("【机构净卖出 Top10】")
print(f"{'代码':<8} {'名称':<10} {'收盘价':>8} {'涨跌幅':>8} {'净卖额':>10} {'换手率':>7} {'上榜原因'}")
print("-" * 90)
for _, row in df_sell.iterrows():
code = row['代码']
name = row['名称']
close = row['收盘价']
chg = row['涨跌幅']
net = row['龙虎榜净买额'] # 负数
turnover = row['换手率']
reason = row['上榜原因'][:20] if pd.notna(row['上榜原因']) else ''
arrow = '🔴' if chg > 0 else '🟢'
print(f"{arrow}{code:<6} {name:<10} {close:>8.2f} {chg:>+7.2f}% {format_amount(net):>10} {turnover:>6.1f}% {reason}")
print()
if __name__ == "__main__":
print_lhb()
FILE:scripts/macro_snapshot.py
"""
A股宏观快照
优先展示 SHIBOR、LPR、中美10Y利差、北向资金和新增人民币贷款。
"""
try:
import akshare as ak
except ImportError:
ak = None
import contextlib
import io
MACRO_NOTES = {
"SHIBOR": {"desc": "银行间借贷成本，衡量流动性松紧", "us_equivalent": "≈ SOFR"},
"LPR": {"desc": "贷款市场报价利率，影响房贷与企业融资", "us_equivalent": "≈ Fed Rate"},
"CN_US_10Y_SPREAD": {"desc": "中美10Y国债利差，影响汇率与外资预期", "us_equivalent": "≈ 10Y-2Y利差"},
"NORTHBOUND": {"desc": "北向资金净流入，反映外资配置A股情绪", "us_equivalent": "—"},
"BANK_FINANCING": {"desc": "新增人民币贷款，反映实体融资需求", "us_equivalent": "—"},
}
def extract_shibor_snapshot(df):
row = df.iloc[-1]
return {
"date": str(row["日期"]),
"on": float(row["O/N-定价"]),
"1w": float(row["1W-定价"]),
"1m": float(row["1M-定价"]),
"3m": float(row["3M-定价"]),
}
def extract_lpr_snapshot(df):
row = df.iloc[-1]
return {
"date": str(row["TRADE_DATE"]),
"1y": float(row["LPR1Y"]),
"5y": float(row["LPR5Y"]),
}
def extract_cn_us_spread(df):
row = df.iloc[-1]
china_10y = float(row["中国国债收益率10年"])
us_10y = float(row["美国国债收益率10年"])
return {
"date": str(row["日期"]),
"china_10y": china_10y,
"us_10y": us_10y,
"spread": china_10y - us_10y,
"spread_bp": (china_10y - us_10y) * 100,
}
def extract_northbound_flow(df):
"""
提取北向资金数据。
若最新交易日是今天，标记为 UNAVAILABLE_TODAY（盘中数据未结算）。
若 sum 为 0 且非今天，标记为 UNAVAILABLE_ZERO。
"""
from datetime import date as date_type
today_str = date_type.today().isoformat()
latest_date_raw = df["交易日"].max()
latest_date_str = str(latest_date_raw)
latest = df[(df["交易日"] == latest_date_raw) & (df["资金方向"] == "北向")]
net_buy = float(latest["成交净买额"].sum())
# 今天的数据尚未结算，显示为"盘中，暂无结算数据"
if latest_date_str == today_str and abs(net_buy) < 0.01:
return {
"date": latest_date_str,
"net_buy": None, # None = 盘中未结算，不代表真实 0
"status": "UNAVAILABLE_TODAY",
}
# 历史日期但净买额为 0，可能是数据缺失
if abs(net_buy) < 0.01:
return {
"date": latest_date_str,
"net_buy": 0.0,
"status": "UNAVAILABLE_ZERO",
}
return {
"date": latest_date_str,
"net_buy": net_buy,
"status": "OK",
}
def extract_bank_financing(df):
row = df.iloc[-1]
return {
"date": str(row["日期"]),
"value": float(row["最新值"]),
"pct": float(row["涨跌幅"]),
}
def get_macro_snapshot():
if ak is None:
return {}
snapshot = {}
try:
snapshot["SHIBOR"] = extract_shibor_snapshot(_quiet_call(ak.macro_china_shibor_all))
except Exception:
pass
try:
snapshot["LPR"] = extract_lpr_snapshot(_quiet_call(ak.macro_china_lpr))
except Exception:
pass
try:
snapshot["CN_US_10Y_SPREAD"] = extract_cn_us_spread(_quiet_call(ak.bond_zh_us_rate))
except Exception:
pass
try:
snapshot["NORTHBOUND"] = extract_northbound_flow(_quiet_call(ak.stock_hsgt_fund_flow_summary_em))
except Exception:
pass
try:
snapshot["BANK_FINANCING"] = extract_bank_financing(_quiet_call(ak.macro_china_bank_financing))
except Exception:
pass
return snapshot
def _quiet_call(func):
sink = io.StringIO()
with contextlib.redirect_stdout(sink), contextlib.redirect_stderr(sink):
return func()
def format_macro_snapshot(snapshot):
lines = []
shibor = snapshot.get("SHIBOR")
if shibor:
lines.append(
f"🔹 SHIBOR ({shibor['date']}): O/N {shibor['on']:.3f}% | 1W {shibor['1w']:.3f}% | "
f"1M {shibor['1m']:.3f}% | 3M {shibor['3m']:.3f}%"
)
lpr = snapshot.get("LPR")
if lpr:
lines.append(f"🔹 LPR ({lpr['date']}): 1Y {lpr['1y']:.2f}% | 5Y {lpr['5y']:.2f}%")
spread = snapshot.get("CN_US_10Y_SPREAD")
if spread:
lines.append(
f"🔹 中美10Y利差 ({spread['date']}): 中国 {spread['china_10y']:.2f}% | 美国 {spread['us_10y']:.2f}% | "
f"利差 {spread['spread_bp']:+.0f}bp"
)
northbound = snapshot.get("NORTHBOUND")
if northbound:
status = northbound.get("status", "OK")
if status == "UNAVAILABLE_TODAY":
lines.append(f"⚪ 北向资金 ({northbound['date']}): 净买额 盘中数据未结算（今日收盘后更新）")
elif status == "UNAVAILABLE_ZERO":
lines.append(f"⚪ 北向资金 ({northbound['date']}): 数据暂不可用")
else:
net_buy = northbound["net_buy"]
arrow = "🔴" if net_buy > 0 else "🟢" if net_buy < 0 else "⚪"
lines.append(f"{arrow} 北向资金 ({northbound['date']}): 净买额 {net_buy:+.2f} 亿")
financing = snapshot.get("BANK_FINANCING")
if financing:
lines.append(
f"🔹 新增人民币贷款 ({financing['date']}): {financing['value']:.0f} 亿元 | "
f"涨跌幅 {financing['pct']:+.2f}%"
)
return lines
if __name__ == "__main__":
print("\n=== 🌏 A股宏观快照 ===")
snapshot = get_macro_snapshot()
if not snapshot:
print("宏观数据暂不可用")
else:
for line in format_macro_snapshot(snapshot):
print(line)
print("\n说明:")
print(f"- SHIBOR: {MACRO_NOTES['SHIBOR']['desc']} {MACRO_NOTES['SHIBOR']['us_equivalent']}")
print(f"- LPR: {MACRO_NOTES['LPR']['desc']} {MACRO_NOTES['LPR']['us_equivalent']}")
print(
f"- 中美10Y利差: {MACRO_NOTES['CN_US_10Y_SPREAD']['desc']} "
f"{MACRO_NOTES['CN_US_10Y_SPREAD']['us_equivalent']}"
)
FILE:scripts/quant_analysis.py
"""
A股宏观量化分析骨架
用于整合流动性、政策、资金结构和情绪信息，输出简要结论。
"""
def _score_signal(value):
mapping = {
"看多": 2,
"偏多": 1,
"中性": 0,
"偏空": -1,
"看空": -2,
}
return mapping.get(value, 0)
def summarize_quant_view(liquidity_view, policy_view, structure_view, sentiment_view):
score = (
_score_signal(liquidity_view)
+ _score_signal(policy_view)
+ _score_signal(structure_view)
+ _score_signal(sentiment_view)
)
if score >= 4:
stance = "积极进攻"
elif score >= 1:
stance = "偏积极"
elif score <= -4:
stance = "防守收缩"
elif score <= -1:
stance = "谨慎观望"
else:
stance = "中性平衡"
return {
"liquidity": liquidity_view,
"policy": policy_view,
"structure": structure_view,
"sentiment": sentiment_view,
"score": score,
"stance": stance,
"conclusion": "先定宏观与政策方向，再看结构资金是否共振。",
}
def build_quant_report(
liquidity_view,
policy_view,
structure_view,
sentiment_view,
main_theme,
risk_note,
):
report = summarize_quant_view(
liquidity_view=liquidity_view,
policy_view=policy_view,
structure_view=structure_view,
sentiment_view=sentiment_view,
)
report["main_theme"] = main_theme
report["risk_note"] = risk_note
return report
def format_quant_report(report):
return (
f"综合倾向: {report['stance']} (score={report['score']}) | "
f"流动性:{report['liquidity']} 政策:{report['policy']} "
f"结构:{report['structure']} 情绪:{report['sentiment']} | "
f"主线: {report['main_theme']} | 风险: {report['risk_note']}"
)
if __name__ == "__main__":
sample = build_quant_report(
liquidity_view="偏多",
policy_view="偏多",
structure_view="中性",
sentiment_view="偏多",
main_theme="稳增长与科技主线轮动",
risk_note="高位拥挤板块若放量滞涨，警惕高低切失败",
)
print("A股量化分析")
print(format_quant_report(sample))
FILE:scripts/sector_ranking.py
"""
东方财富板块涨跌排行
行业板块: m:90+t:2
概念板块: m:90+t:3
【双层架构】
1. 主：东方财富 push2 API
2. 备：东方财富网页 HTML 解析（502/超时/网络错误时切换）
"""
import requests
import sys
import time
import io
def _get_via_api(market_type, top, retries):
"""东方财富 API（主）"""
fs = "m:90+t:2" if market_type == "行业板块" else "m:90+t:3"
url = "https://push2.eastmoney.com/api/qt/clist/get"
params = {
"pn": 1, "pz": top, "po": 1, "np": 1,
"fltt": 2, "invt": 2,
"fid": "f3",
"fs": fs,
"fields": "f12,f14,f2,f3,f5,f6,f8,f10,f15,f16,f18"
}
headers = {"Referer": "https://quote.eastmoney.com/"}
for attempt in range(retries):
try:
r = requests.get(url, params=params, headers=headers, timeout=10)
if r.status_code == 502:
if attempt < retries - 1:
time.sleep(2)
continue
break
r.raise_for_status()
return r.json()["data"]["diff"]
except Exception:
if attempt < retries - 1:
time.sleep(2)
continue
raise
return None # API 层失败，返回 None 触发备用
def _get_via_html(market_type, top=30):
"""
东方财富板块排行 HTML 解析（备用）
通过 pandas.read_html 直接解析网页表格，无需浏览器
"""
market_id = "2" if market_type == "行业板块" else "3"
url = f"https://quote.eastmoney.com/center/boardlist.html#行业板块_{market_id}"
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36",
"Referer": "https://quote.eastmoney.com/",
}
r = requests.get(url, headers=headers, timeout=15)
r.encoding = "utf-8"
r.raise_for_status()
import pandas as pd
# 东方财富板块页面含多个表格，找到正确的那个
tables = pd.read_html(io.StringIO(r.text))
for df in tables:
cols = df.columns.tolist()
# 找含"板块名称"或"涨跌幅"的列
if any("板块" in str(c) for c in cols) or any("涨跌幅" in str(c) for c in cols):
# 标准化列名
df.columns = [str(c) for c in df.columns]
# 找涨跌幅列
pct_col = next((c for c in df.columns if "涨跌幅" in c or "涨跌" in c), None)
name_col = next((c for c in df.columns if "板块" in c or "名称" in c), None)
if pct_col and name_col:
result = []
for _, row in df.iterrows():
try:
pct = float(str(row[pct_col]).replace("%", "").replace("+", ""))
name = str(row[name_col])
if name and name != "nan":
result.append({"f14": name, "f3": pct, "f6": 0})
except ValueError:
continue
return result[:top]
return []
def get_sector_ranking(market_type="行业板块", top=20, retries=3):
"""
获取板块涨跌排行。
优先走 API，失败时自动切换 HTML 解析。
返回 (data, error_msg)：有数据时 error_msg 为 None，无数据时返回具体错误原因。
"""
# 尝试 API（主）
data = None
err = None
try:
data = _get_via_api(market_type, top, retries)
except Exception as e:
err = f"东方财富 API 请求失败: {type(e).__name__}"
if data is not None:
return data, None
# API 失败，切换 HTML 备用（等1秒）
time.sleep(1)
try:
html_data = _get_via_html(market_type, top)
if html_data:
return html_data, None
else:
return [], "东方财富 HTML 解析无结果，数据源可能已更新"
except Exception as e:
return [], f"东方财富 API+HTML 均失败: {type(e).__name__}"
def format_sector(s):
pct = s["f3"]
arrow = "🔴" if pct > 0 else "🟢" if pct < 0 else "⚪"
return f"{arrow} {s['f14']:12s} {'+' if pct>=0 else ''}{pct}% 成交额:{s['f6']/1e8:.1f}亿"
if __name__ == "__main__":
market = sys.argv[1] if len(sys.argv) > 1 else "行业板块"
print(f"\n=== {market}涨跌榜 ===")
data, err = get_sector_ranking(market, 30)
if not data:
if err:
print(f"（{err}，请稍后重试）")
else:
print("（板块数据暂时不可用，请稍后重试）")
else:
print("\n【涨幅榜】")
for s in sorted(data, key=lambda x: x["f3"], reverse=True)[:10]:
print(format_sector(s))
print("\n【跌幅榜】")
for s in sorted(data, key=lambda x: x["f3"])[:10]:
print(format_sector(s))
FILE:scripts/sentiment_snapshot.py
"""
A股短线情绪快照
用于根据涨停/跌停/炸板/连板等指标判断市场情绪阶段。
"""
try:
import akshare as ak
except ImportError:
ak = None
import io
import contextlib
from datetime import datetime, timedelta
SENTIMENT_RULES = {
"冰点": {
"position": "轻仓或空仓",
"note": "情绪低迷，优先控制回撤。",
},
"亢奋": {
"position": "重仓主线龙头",
"note": "情绪高涨，聚焦主线辨识度。",
},
"分歧": {
"position": "半仓参与",
"note": "只做高辨识度标的，避免追高。",
},
"修复": {
"position": "轻仓到半仓",
"note": "观察主线回流与承接强度。",
},
}
def _quiet_call(func, *args, **kwargs):
"""静默调用，屏蔽akshare的print输出"""
sink = io.StringIO()
with contextlib.redirect_stdout(sink), contextlib.redirect_stderr(sink):
return func(*args, **kwargs)
def _is_trading_day():
"""简单判断：工作日"""
return datetime.now().weekday() < 5
def get_sentiment_data():
"""
抓取真实情绪数据。
每个字段独立 try-catch，任一失败 data_fetch_success = False。
"""
if ak is None or not _is_trading_day():
return None
data_fetch_success = True
try:
zt_df = _quiet_call(ak.stock_zt_pool_em)
limit_ups = len(zt_df)
except Exception:
limit_ups = 0
data_fetch_success = False
try:
dt_df = _quiet_call(ak.stock_zt_pool_dtgc_em)
limit_downs = len(dt_df)
except Exception:
limit_downs = 0
data_fetch_success = False
try:
zbgc_df = _quiet_call(ak.stock_zt_pool_zbgc_em)
broken_boards = len(zbgc_df)
except Exception:
broken_boards = 0
data_fetch_success = False
# 连板高度
max_board = 0
if limit_ups > 0 and len(zt_df) > 0:
try:
cols = zt_df.columns.tolist()
board_col = None
for c in ["连板数", "连板", "B板次数"]:
if c in cols:
board_col = c
break
if board_col:
max_board = int(zt_df[board_col].max())
except Exception:
max_board = 0
data_fetch_success = False
return {
"limit_ups": limit_ups,
"limit_downs": limit_downs,
"broken_boards": broken_boards,
"max_board": max_board,
"data_fetch_success": data_fetch_success,
}
def classify_market_sentiment(limit_ups, limit_downs, broken_rate, max_board):
if limit_ups < 20 and limit_downs > 10 and max_board < 3:
stage = "冰点"
elif limit_ups > 80 and broken_rate < 15 and max_board > 5:
stage = "亢奋"
elif broken_rate > 25 or (20 <= limit_ups <= 80 and max_board >= 3):
stage = "分歧"
else:
stage = "修复"
return {
"stage": stage,
"position": SENTIMENT_RULES[stage]["position"],
"note": SENTIMENT_RULES[stage]["note"],
"metrics": {
"limit_ups": limit_ups,
"limit_downs": limit_downs,
"broken_rate": broken_rate,
"max_board": max_board,
},
}
def build_sentiment_snapshot(limit_ups=None, limit_downs=None, broken_boards=None, max_board=None):
"""
构建情绪快照。
若数据获取全部失败（data_fetch_success=False），返回专用状态，不输出情绪结论。
"""
data_fetch_success = True
if all(v is None for v in [limit_ups, limit_downs, broken_boards, max_board]):
data = get_sentiment_data()
if data is None:
return {
"stage": "（非交易日）",
"position": "——",
"note": "周末/节假日数据暂停。",
"metrics": {},
"data_unavailable": True,
}
limit_ups = data["limit_ups"]
limit_downs = data["limit_downs"]
broken_boards = data["broken_boards"]
max_board = data["max_board"]
data_fetch_success = data.get("data_fetch_success", True)
# 数据获取有任意一项失败，不输出情绪结论
if not data_fetch_success:
return {
"stage": "（数据获取失败）",
"position": "——",
"note": "情绪数据暂无法获取，请稍后重试。",
"metrics": {},
"data_unavailable": True,
}
total_limit_events = limit_ups + broken_boards
broken_rate = 0.0 if total_limit_events == 0 else broken_boards / total_limit_events * 100
snapshot = classify_market_sentiment(
limit_ups=limit_ups,
limit_downs=limit_downs,
broken_rate=broken_rate,
max_board=max_board,
)
snapshot["metrics"]["broken_boards"] = broken_boards
snapshot["data_unavailable"] = False
return snapshot
def format_sentiment_snapshot(snapshot):
if snapshot.get("data_unavailable"):
return f"情绪阶段: {snapshot['stage']} | {snapshot['note']}"
metrics = snapshot["metrics"]
if not metrics:
return f"情绪阶段: {snapshot['stage']} | {snapshot['note']}"
return (
f"情绪阶段: {snapshot['stage']} | "
f"仓位建议: {snapshot['position']} | "
f"涨停:{metrics['limit_ups']} 跌停:{metrics['limit_downs']} "
f"炸板:{metrics.get('broken_boards',0)} 炸板率:{metrics['broken_rate']:.1f}% "
f"连板高度:{metrics['max_board']} | {snapshot['note']}"
)
if __name__ == "__main__":
print("\n=== 🎯 A股短线情绪快照 ===")
snapshot = build_sentiment_snapshot()
print(format_sentiment_snapshot(snapshot))
FILE:scripts/stock_quote.py
"""
新浪财经个股实时行情
用法: python stock_quote.py sh600519
python stock_quote.py sh600519 sz000001
"""
import requests
import sys
def fetch_stock_quotes(codes):
if isinstance(codes, str):
codes = [codes]
url = f"https://hq.sinajs.cn/list={','.join(codes)}"
headers = {"Referer": "http://finance.sina.com.cn"}
r = requests.get(url, headers=headers, timeout=10)
r.encoding = 'gbk'
results = []
for line in r.text.strip().split('\n'):
if '=' not in line:
continue
name_part, data = line.split('=', 1)
code = name_part.split('_')[-1].replace('"', '')
vals = data.replace('"', '').replace(';', '').split(',')
if len(vals) < 6:
continue
try:
prev = float(vals[2])
price = float(vals[3])
if prev == 0:
continue
pct = (price - prev) / prev * 100
chg = price - prev
high = float(vals[4])
low = float(vals[5])
volume = float(vals[8]) / 1e8 if len(vals) > 8 and vals[8] else 0
results.append({
"code": code,
"name": vals[0],
"price": price,
"chg": chg,
"pct": pct,
"high": high,
"low": low,
"volume": volume,
})
except (ValueError, IndexError):
continue
return results
def format_stock_quote(item):
arrow = "🔴" if item["pct"] > 0 else "🟢" if item["pct"] < 0 else "⚪"
return (
f"{arrow} {item['name']:8s} {item['code']:8s} "
f"现价:{item['price']:.2f} 涨跌:{item['chg']:+.2f}({item['pct']:+.2f}%) "
f"高:{item['high']:.2f} 低:{item['low']:.2f} 量:{item['volume']:.2f}亿"
)
def get_stock_quote(codes):
results = fetch_stock_quotes(codes)
for item in results:
print(format_stock_quote(item))
if __name__ == "__main__":
if len(sys.argv) < 2:
print("用法: python stock_quote.py sh600519 [sz000001 ...]")
sys.exit(1)
codes = sys.argv[1:]
get_stock_quote(codes)
FILE:scripts/zt_pool.py
"""
涨跌停池查询（需要 akshare）
pip install akshare
"""
try:
import akshare as ak
except ImportError:
print("需要安装akshare: pip install akshare")
exit(1)
def get_zt_pool():
print("\n=== 涨停池 ===")
try:
zt = ak.stock_zt_pool_previous_em()
print(f"涨停家数: {len(zt)}")
print("涨停股(代码/名称/涨幅):")
for _, row in zt.head(15).iterrows():
print(f" 🔴 {row.get('代码','?')} {row.get('名称','?')} {row.get('涨幅','?')}%")
except Exception as e:
print(f"涨停池获取失败: {e}")
print("\n=== 跌停池 ===")
try:
dt = ak.stock_zt_pool_dtgc_em()
print(f"跌停家数: {len(dt)}")
print("跌停股(代码/名称/跌幅):")
for _, row in dt.head(15).iterrows():
print(f" 🟢 {row.get('代码','?')} {row.get('名称','?')} {row.get('跌幅','?')}%")
except Exception as e:
print(f"跌停池获取失败: {e}")
if __name__ == "__main__":
get_zt_pool()