| |
| """ |
| A股微博热搜分析 Demo |
| - 微博热搜抓取 |
| - A股行情数据 |
| - 热门讨论股票分析 |
| """ |
|
|
| import json |
| import sys |
| import urllib.request |
| import urllib.error |
| from datetime import datetime |
|
|
| |
| |
| |
| def fetch_weibo_hot(): |
| """获取微博实时热搜""" |
| url = "https://weibo.com/ajax/side/hotSearch" |
| headers = { |
| "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", |
| "Accept": "application/json", |
| "Referer": "https://weibo.com" |
| } |
| |
| try: |
| req = urllib.request.Request(url, headers=headers) |
| with urllib.request.urlopen(req, timeout=10) as resp: |
| data = json.loads(resp.read().decode("utf-8")) |
| |
| hot_list = data.get("data", {}).get("realtime", []) |
| results = [] |
| for item in hot_list[:50]: |
| results.append({ |
| "rank": item.get("rank", 0), |
| "keyword": item.get("word", ""), |
| "hot": item.get("num", 0), |
| "category": item.get("category", ""), |
| "label": item.get("label_name", ""), |
| }) |
| return results |
| except Exception as e: |
| print(f"⚠️ 微博热搜获取失败: {e}", file=sys.stderr) |
| return [] |
|
|
|
|
| def filter_stock_keywords(hot_list): |
| """筛选A股相关关键词""" |
| stock_keywords = [ |
| "A股", "大A", "股市", "股票", "涨停", "跌停", "牛市", "熊市", |
| "基金", "证券", "券商", "上证", "深证", "创业板", "科创板", |
| "涨停板", "跌停板", "打板", "龙头", "妖股", "板块", |
| "利好", "利空", "暴跌", "暴涨", "反弹", "回调", |
| "北向资金", "主力", "游资", "散户", |
| ] |
| |
| |
| stock_related = [] |
| for item in hot_list: |
| keyword = item["keyword"] |
| |
| for kw in stock_keywords: |
| if kw in keyword: |
| item["match_reason"] = f"包含「{kw}」" |
| stock_related.append(item) |
| break |
| else: |
| |
| if any(w in keyword for w in ["股", "涨", "跌", "板", "牛市", "熊市", "基金"]): |
| item["match_reason"] = "模糊匹配" |
| stock_related.append(item) |
| |
| return stock_related |
|
|
|
|
| |
| |
| |
| def fetch_market_overview(): |
| """获取大盘指数""" |
| try: |
| import akshare as ak |
| |
| |
| sh = ak.stock_zh_index_daily(symbol="sh000001") |
| sz = ak.stock_zh_index_daily(symbol="sz399001") |
| cy = ak.stock_zh_index_daily(symbol="sz399006") |
| |
| def get_latest(df, name): |
| last = df.iloc[-1] |
| prev = df.iloc[-2] |
| close = float(last["close"]) |
| prev_close = float(prev["close"]) |
| change_pct = (close - prev_close) / prev_close * 100 |
| return { |
| "name": name, |
| "close": round(close, 2), |
| "change_pct": round(change_pct, 2), |
| } |
| |
| return [ |
| get_latest(sh, "上证指数"), |
| get_latest(sz, "深证成指"), |
| get_latest(cy, "创业板指"), |
| ] |
| except Exception as e: |
| print(f"⚠️ 大盘数据获取失败: {e}", file=sys.stderr) |
| return [] |
|
|
|
|
| def fetch_zt_dt(): |
| """获取涨停/跌停数据""" |
| try: |
| import akshare as ak |
| |
| |
| zt_df = ak.stock_zt_pool_em(date=datetime.now().strftime("%Y%m%d")) |
| zt_list = [] |
| if zt_df is not None and not zt_df.empty: |
| for _, row in zt_df.head(10).iterrows(): |
| zt_list.append({ |
| "code": str(row.get("代码", "")), |
| "name": str(row.get("名称", "")), |
| "change_pct": float(row.get("涨跌幅", 0)), |
| "reason": str(row.get("涨停原因", "")), |
| }) |
| |
| |
| dt_df = ak.stock_zt_pool_dtgc_em(date=datetime.now().strftime("%Y%m%d")) |
| dt_list = [] |
| if dt_df is not None and not dt_df.empty: |
| for _, row in dt_df.head(10).iterrows(): |
| dt_list.append({ |
| "code": str(row.get("代码", "")), |
| "name": str(row.get("名称", "")), |
| "change_pct": float(row.get("涨跌幅", 0)), |
| }) |
| |
| return {"涨停": zt_list, "跌停": dt_list} |
| except Exception as e: |
| print(f"⚠️ 涨停/跌停数据获取失败: {e}", file=sys.stderr) |
| return {"涨停": [], "跌停": []} |
|
|
|
|
| def fetch_hot_sectors(): |
| """获取热门板块""" |
| try: |
| import akshare as ak |
| |
| df = ak.stock_board_industry_name_em() |
| if df is not None and not df.empty: |
| sectors = [] |
| for _, row in df.head(10).iterrows(): |
| sectors.append({ |
| "name": str(row.get("板块名称", "")), |
| "change_pct": float(row.get("涨跌幅", 0)), |
| "leader": str(row.get("领涨股票", "")), |
| }) |
| return sectors |
| return [] |
| except Exception as e: |
| print(f"⚠️ 板块数据获取失败: {e}", file=sys.stderr) |
| return [] |
|
|
|
|
| |
| |
| |
| def analyze_correlation(hot_list, zt_dt, sectors): |
| """分析热搜与行情的关联""" |
| analysis = { |
| "hot_stock_mentions": [], |
| "potential_analysis": [], |
| } |
| |
| |
| stock_names_in_hot = [] |
| for item in hot_list: |
| keyword = item["keyword"] |
| |
| if 2 <= len(keyword) <= 6 and not any(w in keyword for w in ["A股", "大A", "股市"]): |
| stock_names_in_hot.append(keyword) |
| |
| analysis["hot_stock_mentions"] = stock_names_in_hot[:10] |
| |
| |
| zt_names = [item["name"] for item in zt_dt.get("涨停", [])] |
| mentioned_and_zt = [name for name in stock_names_in_hot if name in zt_names] |
| |
| if mentioned_and_zt: |
| analysis["potential_analysis"].append( |
| f"🔥 以下股票同时出现在微博热搜和涨停板: {', '.join(mentioned_and_zt)}" |
| ) |
| |
| |
| hot_sector_keywords = [item["keyword"] for item in hot_list if "板块" in item["keyword"]] |
| if hot_sector_keywords: |
| analysis["potential_analysis"].append( |
| f"📊 热搜板块关键词: {', '.join(hot_sector_keywords)}" |
| ) |
| |
| return analysis |
|
|
|
|
| |
| |
| |
| def generate_report(hot_list, stock_hot, market, zt_dt, sectors, correlation): |
| """生成分析报告""" |
| now = datetime.now().strftime("%Y-%m-%d %H:%M") |
| |
| lines = [] |
| lines.append(f"📊 A股微博热搜分析报告") |
| lines.append(f"⏰ {now}") |
| lines.append("=" * 50) |
| |
| |
| lines.append("") |
| lines.append("📈 【大盘概览】") |
| lines.append("-" * 40) |
| if market: |
| for m in market: |
| arrow = "🔴" if m["change_pct"] < 0 else "🟢" |
| lines.append(f" {arrow} {m['name']}: {m['close']} ({m['change_pct']:+.2f}%)") |
| else: |
| lines.append(" 数据获取中...") |
| |
| |
| lines.append("") |
| lines.append("🔥 【微博A股热搜】") |
| lines.append("-" * 40) |
| if stock_hot: |
| for item in stock_hot[:15]: |
| hot_num = item["hot"] |
| if hot_num >= 1000000: |
| hot_str = f"{hot_num/10000:.0f}万" |
| elif hot_num >= 10000: |
| hot_str = f"{hot_num/10000:.1f}万" |
| else: |
| hot_str = str(hot_num) |
| lines.append(f" #{item['keyword']}# 🔥{hot_str} ({item['match_reason']})") |
| else: |
| lines.append(" 暂无A股相关热搜") |
| |
| |
| lines.append("") |
| lines.append("🟢 【涨停板 TOP10】") |
| lines.append("-" * 40) |
| if zt_dt.get("涨停"): |
| for item in zt_dt["涨停"]: |
| lines.append(f" {item['name']}({item['code']}) +{item['change_pct']:.1f}% {item.get('reason', '')}") |
| else: |
| lines.append(" 数据获取中...") |
| |
| |
| lines.append("") |
| lines.append("🔴 【跌停板 TOP10】") |
| lines.append("-" * 40) |
| if zt_dt.get("跌停"): |
| for item in zt_dt["跌停"]: |
| lines.append(f" {item['name']}({item['code']}) {item['change_pct']:.1f}%") |
| else: |
| lines.append(" 无跌停或数据获取中...") |
| |
| |
| lines.append("") |
| lines.append("📊 【热门板块 TOP10】") |
| lines.append("-" * 40) |
| if sectors: |
| for item in sectors: |
| arrow = "🔴" if item["change_pct"] < 0 else "🟢" |
| lines.append(f" {arrow} {item['name']}: {item['change_pct']:+.2f}% 领涨: {item['leader']}") |
| else: |
| lines.append(" 数据获取中...") |
| |
| |
| lines.append("") |
| lines.append("🔗 【热搜 vs 行情 关联分析】") |
| lines.append("-" * 40) |
| if correlation["hot_stock_mentions"]: |
| lines.append(f" 💬 热搜提及个股: {', '.join(correlation['hot_stock_mentions'][:5])}") |
| if correlation["potential_analysis"]: |
| for note in correlation["potential_analysis"]: |
| lines.append(f" {note}") |
| if not correlation["potential_analysis"] and not correlation["hot_stock_mentions"]: |
| lines.append(" 暂无明显关联") |
| |
| lines.append("") |
| lines.append("=" * 50) |
| lines.append("📌 数据来源: 微博热搜 + AKShare (东方财富)") |
| lines.append("⚠️ 仅供参考，不构成投资建议") |
| |
| return "\n".join(lines) |
|
|
|
|
| |
| |
| |
| def main(): |
| json_mode = "--json" in sys.argv |
| |
| print("🔍 正在抓取数据...\n", file=sys.stderr) |
| |
| |
| print(" 📱 微博热搜...", file=sys.stderr) |
| hot_list = fetch_weibo_hot() |
| stock_hot = filter_stock_keywords(hot_list) |
| |
| |
| print(" 📈 大盘指数...", file=sys.stderr) |
| market = fetch_market_overview() |
| |
| |
| print(" 🟢 涨停/跌停...", file=sys.stderr) |
| zt_dt = fetch_zt_dt() |
| |
| |
| print(" 📊 热门板块...", file=sys.stderr) |
| sectors = fetch_hot_sectors() |
| |
| |
| print(" 🔗 关联分析...", file=sys.stderr) |
| correlation = analyze_correlation(stock_hot, zt_dt, sectors) |
| |
| |
| if json_mode: |
| result = { |
| "timestamp": datetime.now().isoformat(), |
| "weibo_hot_total": len(hot_list), |
| "stock_hot": stock_hot, |
| "market": market, |
| "zt_dt": zt_dt, |
| "sectors": sectors, |
| "correlation": correlation, |
| } |
| print(json.dumps(result, ensure_ascii=False, indent=2)) |
| else: |
| report = generate_report(hot_list, stock_hot, market, zt_dt, sectors, correlation) |
| print(report) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
