114 lines
3.8 KiB
Python
114 lines
3.8 KiB
Python
"""Fetch market data: major indices, crypto via public APIs (no API key needed)."""
|
|
import json, os, requests
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
OUTPUT = "/var/www/nav/data/market.json"
|
|
NAV_OUTPUT = "/var/www/nav/data/dashboard_market.json"
|
|
tz = timezone(timedelta(hours=8))
|
|
|
|
# Crypto via CoinGecko (free, no key)
|
|
COINGECKO_URL = "https://api.coingecko.com/api/v3/simple/price?ids=bitcoin,ethereum,solana,binancecoin,ripple&vs_currencies=usd&include_24hr_change=true"
|
|
|
|
# A-share indices via Tencent (free, no key)
|
|
TENCENT_URL = "https://qt.gtimg.cn/q=sh000001,sz399001,sz399006"
|
|
|
|
|
|
def fetch_crypto():
|
|
try:
|
|
resp = requests.get(COINGECKO_URL, timeout=10)
|
|
data = resp.json()
|
|
names = {"bitcoin": "BTC", "ethereum": "ETH", "solana": "SOL", "binancecoin": "BNB", "ripple": "XRP"}
|
|
items = []
|
|
for coin_id, symbol in names.items():
|
|
if coin_id not in data:
|
|
continue
|
|
price = data[coin_id]["usd"]
|
|
change = data[coin_id].get("usd_24h_change", 0) or 0
|
|
items.append({
|
|
"type": "crypto",
|
|
"symbol": symbol,
|
|
"price": f"${price:,.2f}",
|
|
"change_pct": f"{abs(change):.2f}%",
|
|
"is_up": change >= 0,
|
|
})
|
|
return items
|
|
except Exception as e:
|
|
print(f" Crypto error: {e}")
|
|
return []
|
|
|
|
|
|
def fetch_cn_indices():
|
|
"""Fetch A-share indices via Tencent qt API (free, no key).
|
|
Field indices: name=1, price=3, timestamp=30, change=31, change_pct=32.
|
|
"""
|
|
try:
|
|
resp = requests.get(TENCENT_URL, timeout=10)
|
|
resp.encoding = "gbk"
|
|
text = resp.text
|
|
items = []
|
|
for line in text.strip().split(";"):
|
|
line = line.strip()
|
|
if not line or "~" not in line:
|
|
continue
|
|
data_str = line.split('"')[1] if '"' in line else ""
|
|
if not data_str:
|
|
continue
|
|
fields = data_str.split("~")
|
|
if len(fields) < 33:
|
|
continue
|
|
name = fields[1]
|
|
price = fields[3]
|
|
change = fields[31]
|
|
change_pct = fields[32]
|
|
try:
|
|
price_f = float(price)
|
|
change_f = float(change)
|
|
pct_f = float(change_pct)
|
|
except (ValueError, TypeError):
|
|
continue
|
|
items.append({
|
|
"type": "cn_index",
|
|
"symbol": name,
|
|
"price": f"{price_f:.2f}",
|
|
"change_pct": f"{abs(pct_f):.2f}%",
|
|
"is_up": change_f >= 0,
|
|
})
|
|
return items
|
|
except Exception as e:
|
|
print(f" CN indices error: {e}")
|
|
return []
|
|
|
|
|
|
def fetch_market():
|
|
crypto = fetch_crypto()
|
|
cn_indices = fetch_cn_indices()
|
|
|
|
# Full data for brief page
|
|
result = {
|
|
"updated_at": datetime.now(tz).strftime("%Y-%m-%d %H:%M"),
|
|
"crypto": crypto,
|
|
"cn_indices": cn_indices,
|
|
}
|
|
|
|
os.makedirs(os.path.dirname(OUTPUT), exist_ok=True)
|
|
with open(OUTPUT, "w", encoding="utf-8") as f:
|
|
json.dump(result, f, ensure_ascii=False, indent=2)
|
|
|
|
# Lightweight data for nav dashboard (no icons, plain names)
|
|
nav_data = {"eth_price": 0, "eth_change_pct": "0%", "eth_is_up": True}
|
|
for c in crypto:
|
|
if c["symbol"] == "ETH":
|
|
nav_data["eth_price"] = c["price"]
|
|
nav_data["eth_change_pct"] = c["change_pct"]
|
|
nav_data["eth_is_up"] = c["is_up"]
|
|
with open(NAV_OUTPUT, "w", encoding="utf-8") as f:
|
|
json.dump(nav_data, f, ensure_ascii=False, indent=2)
|
|
|
|
print(f"Crypto: {len(crypto)} | CN indices: {len(cn_indices)} -> {OUTPUT}")
|
|
return result
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print(f"Fetching market data at {datetime.now(tz).strftime('%Y-%m-%d %H:%M')}")
|
|
fetch_market()
|