nanocode/TradingAgents/tradingagents/dataflows/akshare_news.py

"""
akshare news adapter for TradingAgents.
Replaces yfinance_news for company and global news sourcing.

Uses:
  - stock_news_em    → individual stock news (East Money)
  - news_cctv        → CCTV daily news (macro/global)
  - news_economic_baidu → economic calendar/events
"""

from datetime import datetime, timedelta
from typing import Optional
import os
import pandas as pd
from dateutil.relativedelta import relativedelta

# Suppress akshare's internal tqdm progress bars
os.environ.setdefault('TQDM_DISABLE', '1')

import akshare as ak


# ═══════════════════════════════════════════════════════════════
# Symbol detection
# ═══════════════════════════════════════════════════════════════

def _normalize_ticker(ticker: str) -> str:
    """Normalize ticker for akshare news API.

    stock_news_em accepts: A-share 6-digit codes, or US tickers as-is.
    """
    sym = ticker.strip().upper()
    if sym.isdigit() and len(sym) == 6:
        return sym
    return sym


# ═══════════════════════════════════════════════════════════════
# Company news
# ═══════════════════════════════════════════════════════════════

def get_news(
    ticker: str,
    start_date: str,
    end_date: str,
) -> str:
    """Get company-specific news from East Money.

    Args:
        ticker: Stock symbol (A-share code or US ticker).
        start_date: YYYY-MM-DD format.
        end_date: YYYY-MM-DD format.

    Returns:
        Markdown string containing news items.
    """
    try:
        sym = _normalize_ticker(ticker)
        df = ak.stock_news_em(symbol=sym)

        # Filter by date range
        df["发布时间"] = pd.to_datetime(df["发布时间"])
        mask = (df["发布时间"] >= pd.Timestamp(start_date)) & (
            df["发布时间"] <= pd.Timestamp(end_date) + pd.Timedelta(days=1)
        )
        df = df[mask]

        if df.empty:
            return f"No news found for {ticker} between {start_date} and {end_date}."

        # Select and format key columns
        cols = [c for c in ["发布时间", "新闻标题", "新闻内容", "文章来源"] if c in df.columns]
        result = df[cols].head(30)  # Limit to 30 most recent
        result["发布时间"] = result["发布时间"].astype(str)
        return result.to_markdown(index=False)

    except Exception as e:
        return f"[akshare] Error fetching news for {ticker}: {e}"


# ═══════════════════════════════════════════════════════════════
# Global / macro news
# ═══════════════════════════════════════════════════════════════

def get_global_news(
    curr_date: str,
    look_back_days: int = 7,
    limit: int = 20,
) -> str:
    """Get global macroeconomic news from multiple Chinese news sources.

    Aggregates:
      - CCTV daily news (political/economic headlines)
      - Baidu economic calendar (economic events/data releases)

    Args:
        curr_date: Current reference date (YYYY-MM-DD).
        look_back_days: Days to look back.
        limit: Maximum total news items to return.

    Returns:
        Markdown string with global news and economic events.
    """
    curr_dt = datetime.strptime(curr_date, "%Y-%m-%d")
    sections = []

    # — CCTV news —
    try:
        cctv_parts = []
        for i in range(look_back_days + 1):
            day = curr_dt - relativedelta(days=i)
            day_str = day.strftime("%Y%m%d")
            try:
                df_day = ak.news_cctv(date=day_str)
                if df_day is not None and not df_day.empty:
                    for _, row in df_day.head(10).iterrows():
                        if len(cctv_parts) >= limit:
                            break
                        title = str(row.get("title", "")) if "title" in df_day.columns else ""
                        content = str(row.get("content", "")) if "content" in df_day.columns else ""
                        brief = content[:120] if content else ""
                        cctv_parts.append(f"- [{day_str}] **{title}** — {brief}")
                if len(cctv_parts) >= limit:
                    break
            except Exception:
                continue  # skip days with no data

        if cctv_parts:
            sections.append("## CCTV News (央视新闻)\n\n" + "\n".join(cctv_parts[:limit]))
    except Exception as e:
        sections.append(f"## CCTV News\n\n*Error fetching CCTV news: {e}*")

    # — Economic calendar —
    try:
        econ_parts = []
        for i in range(look_back_days + 1):
            day = curr_dt - relativedelta(days=i)
            day_str = day.strftime("%Y%m%d")
            try:
                df_econ = ak.news_economic_baidu(date=day_str)
                if df_econ is not None and not df_econ.empty:
                    for _, row in df_econ.head(10).iterrows():
                        if len(econ_parts) >= limit:
                            break
                        time_val = str(row.get("time", ""))
                        country = str(row.get("country", "")) if "country" in df_econ.columns else ""
                        event = str(row.get("event", "")) if "event" in df_econ.columns else ""
                        importance = str(row.get("importance", "")) if "importance" in df_econ.columns else ""
                        line = f"- **{country}** {importance} — {event} ({time_val})"
                        econ_parts.append(line)
                if len(econ_parts) >= limit:
                    break
            except Exception:
                continue

        if econ_parts:
            sections.append("## Economic Calendar (经济日历)\n\n" + "\n".join(econ_parts[:limit]))
    except Exception as e:
        sections.append(f"## Economic Calendar\n\n*Error fetching economic calendar: {e}*")

    if not sections:
        return f"No global news found for the period ending {curr_date}."

    header = f"**Global news and economic events from {curr_dt - relativedelta(days=look_back_days):%Y-%m-%d} to {curr_date}**\n\n"
    return header + "\n\n".join(sections)