scripts/demo.py

#!/usr/bin/env python3
"""
A股微博热搜分析 Demo
- 微博热搜抓取
- A股行情数据
- 热门讨论股票分析
"""

import json
import sys
import urllib.request
import urllib.error
from datetime import datetime

# ============================================================
# 1. 微博热搜抓取
# ============================================================
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]:  # 取前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", "股市", "股票", "涨停", "跌停", "牛市", "熊市",
        "基金", "证券", "券商", "上证", "深证", "创业板", "科创板",
        "涨停板", "跌停板", "打板", "龙头", "妖股", "板块",
        "利好", "利空", "暴跌", "暴涨", "反弹", "回调",
        "北向资金", "主力", "游资", "散户",
    ]
    
    # 额外：提取可能的个股名称（2-4个汉字）
    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


# ============================================================
# 2. A股行情数据 (AKShare)
# ============================================================
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 []


# ============================================================
# 3. 关联分析
# ============================================================
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"]
        # 简单提取：如果关键词包含具体股票名（2-4字中文）
        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


# ============================================================
# 4. 报告生成
# ============================================================
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("  数据获取中...")
    
    # 微博A股热搜
    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)


# ============================================================
# Main
# ============================================================
def main():
    json_mode = "--json" in sys.argv
    
    print("🔍 正在抓取数据...\n", file=sys.stderr)
    
    # 1. 微博热搜
    print("  📱 微博热搜...", file=sys.stderr)
    hot_list = fetch_weibo_hot()
    stock_hot = filter_stock_keywords(hot_list)
    
    # 2. 大盘指数
    print("  📈 大盘指数...", file=sys.stderr)
    market = fetch_market_overview()
    
    # 3. 涨停/跌停
    print("  🟢 涨停/跌停...", file=sys.stderr)
    zt_dt = fetch_zt_dt()
    
    # 4. 热门板块
    print("  📊 热门板块...", file=sys.stderr)
    sectors = fetch_hot_sectors()
    
    # 5. 关联分析
    print("  🔗 关联分析...", file=sys.stderr)
    correlation = analyze_correlation(stock_hot, zt_dt, sectors)
    
    # 6. 输出
    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()