Findata Toolkit Cn

# 金融数据工具包 — A股市场数据源配置
# ====================================
# 所有主要数据源免费，无需API密钥。

china_market:
  primary:
    stock_data: "akshare"           # AKShare — 免费，无需密钥

  # 可选：在 https://tushare.pro/register 注册
  # export TUSHARE_TOKEN=your_token_here
  optional:
    tushare_token: "${TUSHARE_TOKEN}"

# 速率限制设置（每秒请求数）
rate_limits:
  akshare: 5

Copyright 2025 FinoGeeks Technology Ltd

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

# 金融数据工具包 — A股市场依赖
# 安装: pip install -r requirements.txt

# === 核心数据获取 ===
akshare>=1.12.0           # A股数据（无需API密钥）

# === 数据处理 ===
pandas>=2.0.0
numpy>=1.24.0

# === 输出 ===
tabulate>=0.9.0           # 表格美化输出

# 金融数据工具包 — 通用工具

"""
金融数据工具包配置管理器
加载数据源配置和API密钥。
"""
import os
import json
from pathlib import Path


# 基础路径
SCRIPT_DIR = Path(__file__).resolve().parent.parent
PROJECT_DIR = SCRIPT_DIR.parent
CONFIG_DIR = PROJECT_DIR / "config"


def get_config() -> dict:
    """从 YAML 加载配置或返回默认值。"""
    try:
        import yaml
        config_path = CONFIG_DIR / "data_sources.yaml"
        if config_path.exists():
            with open(config_path) as f:
                cfg = yaml.safe_load(f)
            _resolve_env_vars(cfg)
            return cfg
    except ImportError:
        pass

    # 回退默认值（无需外部配置）
    return {
        "china_market": {
            "primary": {
                "stock_data": "akshare",
            },
            "optional": {
                "tushare_token": os.getenv("TUSHARE_TOKEN"),
            },
        },
        "rate_limits": {
            "akshare": 5,
        },
    }


def _resolve_env_vars(obj):
    """递归解析配置值中的 ${ENV_VAR} 引用。"""
    if isinstance(obj, dict):
        for k, v in obj.items():
            if isinstance(v, str) and v.startswith("${") and v.endswith("}"):
                env_key = v[2:-1]
                obj[k] = os.getenv(env_key)
            elif isinstance(v, (dict, list)):
                _resolve_env_vars(v)
    elif isinstance(obj, list):
        for i, v in enumerate(obj):
            if isinstance(v, str) and v.startswith("${") and v.endswith("}"):
                env_key = v[2:-1]
                obj[i] = os.getenv(env_key)
            elif isinstance(v, (dict, list)):
                _resolve_env_vars(v)

"""
金融数据工具包共享工具
所有脚本通用函数。
"""
import json
import sys
import time
import functools
from datetime import datetime, date
from typing import Any


# ---------------------------------------------------------------------------
# 输出助手
# ---------------------------------------------------------------------------

class JSONEncoder(json.JSONEncoder):
    """自定义编码器，处理日期、numpy 类型等。"""

    def default(self, obj):
        if isinstance(obj, (datetime, date)):
            return obj.isoformat()
        try:
            import numpy as np
            if isinstance(obj, (np.integer,)):
                return int(obj)
            if isinstance(obj, (np.floating,)):
                return float(obj)
            if isinstance(obj, np.ndarray):
                return obj.tolist()
            if isinstance(obj, np.bool_):
                return bool(obj)
        except ImportError:
            pass
        try:
            import pandas as pd
            if isinstance(obj, pd.Timestamp):
                return obj.isoformat()
            if pd.isna(obj):
                return None
        except ImportError:
            pass
        return super().default(obj)


def output_json(data: Any, pretty: bool = True) -> str:
    """将数据序列化为 JSON 并输出到标准输出。"""
    text = json.dumps(data, cls=JSONEncoder, indent=2 if pretty else None,
                      ensure_ascii=False)
    print(text)
    return text


def output_table(headers: list[str], rows: list[list], title: str = ""):
    """输出格式化文本表格。"""
    try:
        from tabulate import tabulate
        if title:
            print(f"\n{'='*60}")
            print(f"  {title}")
            print(f"{'='*60}")
        print(tabulate(rows, headers=headers, tablefmt="pipe",
                        floatfmt=".2f"))
    except ImportError:
        if title:
            print(f"\n--- {title} ---")
        print(" | ".join(headers))
        print("-" * (len(" | ".join(headers))))
        for row in rows:
            print(" | ".join(str(c) for c in row))


def error_exit(message: str, code: int = 1):
    """输出错误信息并退出。"""
    print(json.dumps({"error": message}), file=sys.stderr)
    sys.exit(code)


# ---------------------------------------------------------------------------
# 速率限制
# ---------------------------------------------------------------------------

def rate_limit(calls_per_second: float = 2.0):
    """限速装饰器。"""
    min_interval = 1.0 / calls_per_second

    def decorator(func):
        last_call = [0.0]

        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_call[0]
            if elapsed < min_interval:
                time.sleep(min_interval - elapsed)
            last_call[0] = time.time()
            return func(*args, **kwargs)

        return wrapper
    return decorator


# ---------------------------------------------------------------------------
# 安全数值助手
# ---------------------------------------------------------------------------

def safe_div(a, b, default=None):
    """安全除法，b 为 0 或任一值为 None 时返回 default。"""
    if a is None or b is None:
        return default
    try:
        if float(b) == 0:
            return default
        return float(a) / float(b)
    except (ValueError, TypeError):
        return default


def safe_float(val, default=None):
    """安全转换为 float，失败时返回 default。处理 %、亿、万 等后缀。

    Note: '%' is stripped but the value is NOT divided by 100 — e.g. "15.3%"
    returns 15.3.  Callers that need a fraction should divide by 100 themselves.
    """
    if val is None or val is False:  # AKShare may return False for missing data
        return default
    import math
    if isinstance(val, (int, float)):
        if math.isnan(val) or math.isinf(val):
            return default
        return float(val)
    try:
        s = str(val).strip()
        if not s:
            return default
        multiplier = 1.0
        if s.endswith('%'):
            s = s[:-1]
        elif s.endswith('亿'):
            s = s[:-1]
            multiplier = 1e8
        elif s.endswith('万'):
            s = s[:-1]
            multiplier = 1e4
        result = float(s) * multiplier
        if math.isnan(result) or math.isinf(result):
            return default
        return result
    except (ValueError, TypeError):
        return default


def pct(val, decimals=2):
    """将小数格式化为百分比字符串。"""
    if val is None:
        return "N/A"
    return f"{val * 100:.{decimals}f}%"

#!/usr/bin/env python3
"""
China Macro Economic Data Fetcher
====================================
Fetch A-share macro indicators using AKShare (no API key required).

Covers: LPR, CPI/PPI, GDP, PMI, social financing, money supply, northbound flow.

Usage:
    python macro_data.py --dashboard                  # All key indicators
    python macro_data.py --rates                      # Interest rates (LPR, MLF)
    python macro_data.py --inflation                  # CPI/PPI data
    python macro_data.py --pmi                        # PMI data
    python macro_data.py --social-financing            # Social financing
    python macro_data.py --cycle                      # Business cycle assessment
"""
import argparse
import sys
from datetime import datetime
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from common.utils import output_json, safe_float, error_exit


def _direction(values: list, lookback: int = 6) -> str:
    """Determine trend direction from a list of numeric values."""
    valid = [v for v in values if v is not None]
    if len(valid) < 2:
        return "insufficient_data"
    recent = valid[-lookback:]
    if len(recent) < 2:
        return "insufficient_data"
    change = (recent[-1] - recent[0]) / abs(recent[0]) if recent[0] != 0 else 0
    if change > 0.03:
        return "rising"
    elif change < -0.03:
        return "falling"
    return "stable"


# ---------------------------------------------------------------------------
# Interest Rates
# ---------------------------------------------------------------------------

def fetch_rates() -> dict:
    """Fetch Chinese interest rate data (LPR, MLF, etc.)."""
    import akshare as ak

    result = {}

    # LPR (Loan Prime Rate)
    try:
        df = ak.macro_china_lpr()
        if df is not None and not df.empty:
            recent = df.tail(12)
            lpr_1y = []
            lpr_5y = []
            for _, row in recent.iterrows():
                lpr_1y.append({
                    "date": str(row.get("TRADE_DATE", "")),
                    "value": safe_float(row.get("LPR1Y")),
                })
                lpr_5y.append({
                    "date": str(row.get("TRADE_DATE", "")),
                    "value": safe_float(row.get("LPR5Y")),
                })
            result["lpr_1y"] = {
                "latest": lpr_1y[-1]["value"] if lpr_1y else None,
                "direction": _direction([e["value"] for e in lpr_1y]),
                "series": lpr_1y,
            }
            result["lpr_5y"] = {
                "latest": lpr_5y[-1]["value"] if lpr_5y else None,
                "direction": _direction([e["value"] for e in lpr_5y]),
                "series": lpr_5y,
            }
    except Exception as e:
        result["lpr"] = {"error": str(e)}

    # Shibor
    try:
        df = ak.rate_interbank(market="上海银行间同业拆放利率(Shibor)", symbol="隔夜",
                               indicator="利率")
        if df is not None and not df.empty:
            recent = df.tail(30)
            records = []
            for _, row in recent.iterrows():
                records.append({
                    "date": str(row.iloc[0]) if len(row) > 0 else "",
                    "value": safe_float(row.iloc[1]) if len(row) > 1 else None,
                })
            result["shibor_overnight"] = {
                "latest": records[-1]["value"] if records else None,
                "direction": _direction([e["value"] for e in records]),
            }
    except Exception:
        result["shibor_overnight"] = {"note": "Data not available"}

    return result


# ---------------------------------------------------------------------------
# Inflation
# ---------------------------------------------------------------------------

def fetch_inflation() -> dict:
    """Fetch Chinese CPI and PPI data."""
    import akshare as ak

    result = {}

    # CPI
    try:
        df = ak.macro_china_cpi_monthly()
        if df is not None and not df.empty:
            # Columns: ['商品', '日期', '今值', '预测值', '前值']
            # Chronological order (oldest first), so .tail(12) is correct
            recent = df.tail(12)
            records = []
            for _, row in recent.iterrows():
                records.append({
                    "date": str(row.iloc[1]),
                    "cpi_yoy": safe_float(row.iloc[2]) if len(row) > 2 else None,
                })
            result["cpi"] = {
                "latest": records[-1]["cpi_yoy"] if records else None,
                "direction": _direction([e["cpi_yoy"] for e in records]),
                "series": records,
            }
    except Exception as e:
        result["cpi"] = {"error": str(e)}

    # PPI
    try:
        df = ak.macro_china_ppi()
        if df is not None and not df.empty:
            # Columns: ['月份', '当月', '当月同比增长', '累计']
            # Reverse chronological order (newest first)
            recent = df.head(12).iloc[::-1]  # take newest 12, reverse to ascending
            records = []
            for _, row in recent.iterrows():
                records.append({
                    "date": str(row.iloc[0]),
                    "ppi_yoy": safe_float(row.iloc[2]) if len(row) > 2 else None,
                })
            result["ppi"] = {
                "latest": records[-1]["ppi_yoy"] if records else None,
                "direction": _direction([e["ppi_yoy"] for e in records]),
                "series": records,
            }
    except Exception as e:
        result["ppi"] = {"error": str(e)}

    return result


# ---------------------------------------------------------------------------
# PMI
# ---------------------------------------------------------------------------

def fetch_pmi() -> dict:
    """Fetch China PMI data."""
    import akshare as ak

    result = {}

    try:
        df = ak.macro_china_pmi()
        if df is not None and not df.empty:
            # macro_china_pmi() returns reverse chronological order (newest first)
            recent = df.head(12).iloc[::-1]  # take newest 12, reverse to ascending
            records = []
            for _, row in recent.iterrows():
                records.append({
                    "date": str(row.iloc[0]),
                    "manufacturing_pmi": safe_float(row.iloc[1]) if len(row) > 1 else None,
                    "non_manufacturing_pmi": safe_float(row.iloc[3]) if len(row) > 3 else None,
                })
            mfg_values = [e["manufacturing_pmi"] for e in records]
            result["manufacturing_pmi"] = {
                "latest": records[-1]["manufacturing_pmi"] if records else None,
                "direction": _direction(mfg_values),
                "above_50": (records[-1]["manufacturing_pmi"] or 0) > 50 if records else None,
                "interpretation": (
                    "Above 50 — manufacturing expanding"
                    if records and (records[-1]["manufacturing_pmi"] or 0) > 50
                    else "Below 50 — manufacturing contracting"
                ),
                "series": records,
            }
    except Exception as e:
        result["manufacturing_pmi"] = {"error": str(e)}

    return result


# ---------------------------------------------------------------------------
# Social Financing
# ---------------------------------------------------------------------------

def fetch_social_financing() -> dict:
    """Fetch China social financing data (社会融资规模)."""
    import akshare as ak

    result = {}

    try:
        df = ak.macro_china_shrzgm()
        if df is not None and not df.empty:
            recent = df.tail(12)
            records = []
            for _, row in recent.iterrows():
                records.append({
                    "date": str(row.iloc[0]),
                    "value": safe_float(row.iloc[1]) if len(row) > 1 else None,
                })
            result["social_financing"] = {
                "latest": records[-1]["value"] if records else None,
                "direction": _direction([e["value"] for e in records]),
                "series": records,
                "interpretation": "Social financing growth indicates credit expansion/contraction",
            }
    except Exception as e:
        result["social_financing"] = {"error": str(e)}

    # Money supply M2
    try:
        df = ak.macro_china_money_supply()
        if df is not None and not df.empty:
            # Columns: ['月份', '货币和准货币(M2)-数量(亿元)', '货币和准货币(M2)-同比增长', ...]
            # Reverse chronological order (newest first)
            recent = df.head(12).iloc[::-1]  # take newest 12, reverse to ascending
            records = []
            for _, row in recent.iterrows():
                records.append({
                    "date": str(row.iloc[0]),
                    "m2_yoy": safe_float(row.iloc[2]) if len(row) > 2 else None,
                })
            result["m2_growth"] = {
                "latest": records[-1]["m2_yoy"] if records else None,
                "direction": _direction([e["m2_yoy"] for e in records]),
                "series": records,
            }
    except Exception as e:
        result["m2_growth"] = {"error": str(e)}

    return result


# ---------------------------------------------------------------------------
# Business Cycle Assessment (China)
# ---------------------------------------------------------------------------

def assess_business_cycle() -> dict:
    """
    Determine current China business cycle phase.
    Uses PMI, CPI, PPI, credit data, and policy signals.
    """
    inflation = fetch_inflation()
    pmi_data = fetch_pmi()
    financing = fetch_social_financing()

    signals = {}

    # PMI signal
    mfg_pmi = pmi_data.get("manufacturing_pmi", {})
    pmi_latest = mfg_pmi.get("latest")
    pmi_dir = mfg_pmi.get("direction", "stable")
    signals["pmi"] = {
        "value": pmi_latest,
        "direction": pmi_dir,
        "expanding": pmi_latest > 50 if pmi_latest else None,
    }

    # Inflation signals
    cpi_latest = inflation.get("cpi", {}).get("latest")
    ppi_latest = inflation.get("ppi", {}).get("latest")
    signals["cpi"] = {"value": cpi_latest, "direction": inflation.get("cpi", {}).get("direction")}
    signals["ppi"] = {"value": ppi_latest, "direction": inflation.get("ppi", {}).get("direction")}

    # Credit signal
    sf = financing.get("social_financing", {})
    sf_dir = sf.get("direction", "stable")
    m2 = financing.get("m2_growth", {})
    m2_dir = m2.get("direction", "stable")
    signals["credit"] = {"social_financing_direction": sf_dir, "m2_direction": m2_dir}

    # Phase determination
    pmi_expanding = pmi_latest and pmi_latest > 50
    pmi_rising = pmi_dir == "rising"
    credit_expanding = sf_dir == "rising" or m2_dir == "rising"

    if pmi_expanding and pmi_rising and credit_expanding:
        phase = "recovery"
        description = "经济复苏期：PMI回升，信用扩张，政策宽松"
        favored = ["消费", "科技", "金融"]
        disfavored = ["公用事业"]
    elif pmi_expanding and not pmi_rising:
        phase = "expansion"
        description = "经济扩张期：PMI维持高位，增长稳定"
        favored = ["制造业", "周期股", "金融"]
        disfavored = ["防御板块"]
    elif not pmi_expanding and ppi_latest and ppi_latest < 0:
        phase = "contraction"
        description = "经济收缩期：PMI低于50，PPI通缩"
        favored = ["消费防御", "公用事业", "高股息"]
        disfavored = ["周期股", "地产"]
    else:
        phase = "transition"
        description = "过渡期：经济信号混合"
        favored = ["均衡配置"]
        disfavored = []

    return {
        "phase": phase,
        "description": description,
        "signals": signals,
        "sector_implications": {
            "favored": favored,
            "disfavored": disfavored,
        },
        "factor_implications": {
            "recovery": "小盘、动量因子占优",
            "expansion": "质量、成长因子占优",
            "contraction": "低波动、红利因子占优",
            "transition": "均衡配置各因子",
        }.get(phase, ""),
    }


# ---------------------------------------------------------------------------
# Dashboard
# ---------------------------------------------------------------------------

def macro_dashboard() -> dict:
    """Comprehensive China macro dashboard."""
    return {
        "timestamp": datetime.now().isoformat(),
        "rates": fetch_rates(),
        "inflation": fetch_inflation(),
        "pmi": fetch_pmi(),
        "social_financing": fetch_social_financing(),
        "business_cycle": assess_business_cycle(),
    }


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(
        description="China Macro Data Fetcher (AKShare, no API key)"
    )
    parser.add_argument("--dashboard", action="store_true", help="Full dashboard")
    parser.add_argument("--rates", action="store_true", help="Interest rates")
    parser.add_argument("--inflation", action="store_true", help="CPI/PPI")
    parser.add_argument("--pmi", action="store_true", help="PMI data")
    parser.add_argument("--social-financing", action="store_true",
                        help="Social financing + M2")
    parser.add_argument("--cycle", action="store_true",
                        help="Business cycle assessment")
    args = parser.parse_args()

    try:
        if args.rates:
            data = fetch_rates()
        elif args.inflation:
            data = fetch_inflation()
        elif args.pmi:
            data = fetch_pmi()
        elif args.social_financing:
            data = fetch_social_financing()
        elif args.cycle:
            data = assess_business_cycle()
        else:
            data = macro_dashboard()

        output_json(data)

    except ImportError:
        error_exit("akshare is required. Install: pip install akshare")
    except Exception as e:
        error_exit(f"Error: {e}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
A-Share Market Stock Data Fetcher
=================================
Fetch A-share stock fundamentals, price history, financial metrics,
and insider trading data using AKShare (no API key required).

Usage:
    python stock_data.py 600519                       # Basic info (Kweichow Moutai)
    python stock_data.py 600519 --metrics             # Full financial metrics
    python stock_data.py 600519 --history             # Price history
    python stock_data.py 600519 --financials          # Financial statements
    python stock_data.py 600519 --insider             # Insider trades
    python stock_data.py 600519 000858 --screen       # Screen with filters
"""
import argparse
import sys
from datetime import datetime, timedelta
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from common.utils import output_json, safe_div, safe_float, error_exit


def _normalize_symbol(symbol: str) -> str:
    """Normalize A-share symbol to 6-digit format."""
    sym = symbol.strip().replace(".SH", "").replace(".SZ", "").replace(".BJ", "")
    return sym.zfill(6)


def _get_exchange_suffix(symbol: str) -> str:
    """Determine exchange from symbol prefix."""
    sym = _normalize_symbol(symbol)
    if sym.startswith(("6",)):
        return "sh"  # Shanghai
    elif sym.startswith(("0", "3")):
        return "sz"  # Shenzhen
    elif sym.startswith(("4", "8")):
        return "bj"  # Beijing Stock Exchange
    return "sh"


def fetch_basic_info(symbols: list[str]) -> list[dict]:
    """Fetch basic company info for A-share stocks."""
    import akshare as ak

    results = []
    for symbol in symbols:
        sym = _normalize_symbol(symbol)
        try:
            # Get real-time quote
            df = ak.stock_individual_info_em(symbol=sym)
            info = {}
            if df is not None and not df.empty:
                for _, row in df.iterrows():
                    key = str(row.iloc[0])
                    val = row.iloc[1]
                    info[key] = val

            results.append({
                "symbol": sym,
                "name": info.get("股票简称", ""),
                "industry": info.get("行业", ""),
                "market_cap": safe_float(info.get("总市值")),
                "circulating_cap": safe_float(info.get("流通市值")),
                "total_shares": safe_float(info.get("总股本")),
                "circulating_shares": safe_float(info.get("流通股")),
                "exchange": _get_exchange_suffix(sym),
                "listing_date": info.get("上市时间", ""),
            })
        except Exception as e:
            results.append({"symbol": sym, "error": str(e)})

    return results


def fetch_financial_metrics(symbol: str) -> dict:
    """
    Fetch comprehensive financial metrics for a single A-share stock.
    Uses AKShare to pull valuation, profitability, leverage, and growth data.
    """
    import akshare as ak

    sym = _normalize_symbol(symbol)

    result = {
        "symbol": sym,
        "name": "",
        "industry": "",
        "current_price": None,
        "market_cap": None,
    }

    # --- Basic info ---
    try:
        df_info = ak.stock_individual_info_em(symbol=sym)
        info = {}
        if df_info is not None and not df_info.empty:
            for _, row in df_info.iterrows():
                info[str(row.iloc[0])] = row.iloc[1]
        result["name"] = info.get("股票简称", "")
        result["industry"] = info.get("行业", "")
        result["market_cap"] = safe_float(info.get("总市值"))
    except Exception:
        pass

    # --- Real-time quote ---
    try:
        df_quote = ak.stock_zh_a_spot_em()
        if df_quote is not None and not df_quote.empty:
            row = df_quote[df_quote["代码"] == sym]
            if not row.empty:
                row = row.iloc[0]
                result["current_price"] = safe_float(row.get("最新价"))
                result["valuation"] = {
                    "pe_ttm": safe_float(row.get("市盈率-动态")),
                    "pb": safe_float(row.get("市净率")),
                    "total_market_cap": safe_float(row.get("总市值")),
                    "circulating_cap": safe_float(row.get("流通市值")),
                }
                result["trading"] = {
                    "change_pct": safe_float(row.get("涨跌幅")),
                    "turnover_rate": safe_float(row.get("换手率")),
                    "volume": safe_float(row.get("成交量")),
                    "amount": safe_float(row.get("成交额")),
                    "amplitude": safe_float(row.get("振幅")),
                }
    except Exception:
        pass

    # --- Financial indicators (profitability, leverage) ---
    try:
        df_fin = ak.stock_financial_abstract_ths(symbol=sym, indicator="按报告期")
        if df_fin is not None and not df_fin.empty:
            latest = df_fin.iloc[-1]
            result["profitability"] = {
                "roe": safe_float(latest.get("净资产收益率")),
                "gross_margin": safe_float(latest.get("销售毛利率")),
                "net_margin": safe_float(latest.get("销售净利率")),
            }
            result["leverage"] = {
                "debt_to_asset_ratio": safe_float(latest.get("资产负债率")),
                "current_ratio": safe_float(latest.get("流动比率")),
                "quick_ratio": safe_float(latest.get("速动比率")),
            }
            result["growth"] = {
                "revenue_growth_yoy": safe_float(latest.get("营业总收入同比增长率")),
                "profit_growth_yoy": safe_float(latest.get("净利润同比增长率")),
            }
            result["per_share"] = {
                "eps": safe_float(latest.get("基本每股收益")),
                "bvps": safe_float(latest.get("每股净资产")),
                "ocf_per_share": safe_float(latest.get("每股经营现金流")),
            }
    except Exception:
        pass

    # --- Dividend data ---
    try:
        df_div = ak.stock_history_dividend_detail(symbol=sym, indicator="分红")
        if df_div is not None and not df_div.empty:
            recent_divs = df_div.head(5)
            dividends = []
            for _, row in recent_divs.iterrows():
                dividends.append({
                    "report_date": str(row.get("公告日期", "")),
                    "dividend_per_share": safe_float(row.get("派息")),
                    "ex_date": str(row.get("除权除息日", "")),
                })
            result["dividends"] = dividends
    except Exception:
        result["dividends"] = []

    return result


def fetch_price_history(symbol: str, period: str = "1y",
                        adjust: str = "qfq") -> dict:
    """
    Fetch historical OHLCV data for an A-share stock.

    Args:
        symbol: A-share stock code (e.g., "600519")
        period: "1m", "3m", "6m", "1y", "2y", "5y", "max"
        adjust: "qfq" (forward-adjusted), "hfq" (backward-adjusted), "" (unadjusted)
    """
    import akshare as ak

    sym = _normalize_symbol(symbol)

    period_map = {
        "1m": 30, "3m": 90, "6m": 180, "1y": 365,
        "2y": 730, "5y": 1825, "max": 7300,
    }
    days = period_map.get(period, 365)
    start = (datetime.now() - timedelta(days=days)).strftime("%Y%m%d")
    end = datetime.now().strftime("%Y%m%d")

    try:
        df = ak.stock_zh_a_hist(
            symbol=sym, period="daily",
            start_date=start, end_date=end,
            adjust=adjust
        )
    except Exception as e:
        return {"symbol": sym, "error": str(e)}

    if df is None or df.empty:
        return {"symbol": sym, "error": "No price data found"}

    records = []
    for _, row in df.iterrows():
        records.append({
            "date": str(row.get("日期", "")),
            "open": safe_float(row.get("开盘")),
            "high": safe_float(row.get("最高")),
            "low": safe_float(row.get("最低")),
            "close": safe_float(row.get("收盘")),
            "volume": safe_float(row.get("成交量")),
            "amount": safe_float(row.get("成交额")),
            "turnover_rate": safe_float(row.get("换手率")),
        })

    return {
        "symbol": sym,
        "period": period,
        "adjust": adjust,
        "data_points": len(records),
        "start_date": records[0]["date"] if records else "",
        "end_date": records[-1]["date"] if records else "",
        "prices": records,
    }


def fetch_financial_statements(symbol: str) -> dict:
    """Fetch income statement, balance sheet, and cash flow for A-share stocks."""
    import akshare as ak

    sym = _normalize_symbol(symbol)

    result = {"symbol": sym}

    # --- Income Statement ---
    try:
        df = ak.stock_financial_report_sina(stock=sym, symbol="利润表")
        if df is not None and not df.empty:
            records = []
            for col in df.columns[:5]:  # Last 5 periods
                period_data = {"period": str(col)}
                for idx in df.index:
                    val = safe_float(df.loc[idx, col])
                    period_data[str(idx)] = val
                records.append(period_data)
            result["income_statement"] = records
    except Exception:
        result["income_statement"] = []

    # --- Balance Sheet ---
    try:
        df = ak.stock_financial_report_sina(stock=sym, symbol="资产负债表")
        if df is not None and not df.empty:
            records = []
            for col in df.columns[:5]:
                period_data = {"period": str(col)}
                for idx in df.index:
                    val = safe_float(df.loc[idx, col])
                    period_data[str(idx)] = val
                records.append(period_data)
            result["balance_sheet"] = records
    except Exception:
        result["balance_sheet"] = []

    # --- Cash Flow Statement ---
    try:
        df = ak.stock_financial_report_sina(stock=sym, symbol="现金流量表")
        if df is not None and not df.empty:
            records = []
            for col in df.columns[:5]:
                period_data = {"period": str(col)}
                for idx in df.index:
                    val = safe_float(df.loc[idx, col])
                    period_data[str(idx)] = val
                records.append(period_data)
            result["cash_flow"] = records
    except Exception:
        result["cash_flow"] = []

    return result


def _get_exchange_prefix(symbol: str) -> str:
    """Return exchange prefix for insider trade filtering (e.g. 'SH688557')."""
    sym = _normalize_symbol(symbol)
    if sym.startswith("6"):
        return "SH" + sym
    elif sym.startswith(("0", "3")):
        return "SZ" + sym
    elif sym.startswith(("4", "8")):
        return "BJ" + sym
    return "SH" + sym


def fetch_insider_trades(symbol: str) -> dict:
    """Fetch insider trading (董监高增减持) data for an A-share stock."""
    import akshare as ak

    sym = _normalize_symbol(symbol)
    prefixed = _get_exchange_prefix(sym)

    try:
        df = ak.stock_inner_trade_xq()
        if df is None or df.empty:
            return {"symbol": sym, "transactions": [], "note": "No insider trades found"}

        # Filter for our target symbol
        df = df[df["股票代码"] == prefixed]
        if df.empty:
            return {"symbol": sym, "transactions": [], "note": "No insider trades found"}

        trades = []
        for _, row in df.iterrows():
            shares_changed = safe_float(row.get("变动股数"))
            if shares_changed is not None and shares_changed > 0:
                change_type = "增持"
            elif shares_changed is not None and shares_changed < 0:
                change_type = "减持"
            else:
                change_type = "未知"

            trades.append({
                "name": str(row.get("变动人", "")),
                "position": str(row.get("董监高职务", "")),
                "relationship": str(row.get("与董监高关系", "")),
                "change_type": change_type,
                "shares_changed": shares_changed,
                "price": safe_float(row.get("成交均价")),
                "shares_after": safe_float(row.get("变动后持股数")),
                "date": str(row.get("变动日期", "")),
            })

        buys = [t for t in trades if t["change_type"] == "增持"]
        sells = [t for t in trades if t["change_type"] == "减持"]

        return {
            "symbol": sym,
            "total_transactions": len(trades),
            "summary": {
                "total_purchases": len(buys),
                "total_sales": len(sells),
                "unique_buyers": len(set(t["name"] for t in buys)),
            },
            "transactions": trades,
        }
    except Exception as e:
        return {"symbol": sym, "error": str(e)}


def fetch_northbound_flow() -> dict:
    """Fetch northbound capital flow data (北向资金/沪深港通)."""
    import akshare as ak

    try:
        df = ak.stock_hsgt_north_net_flow_in_em(symbol="北向")
        if df is None or df.empty:
            return {"error": "No northbound flow data"}

        # Last 30 days
        records = []
        for _, row in df.tail(30).iterrows():
            records.append({
                "date": str(row.get("日期", "")),
                "net_inflow": safe_float(row.get("当日净流入")),
                "sh_connect": safe_float(row.get("沪股通净流入")),
                "sz_connect": safe_float(row.get("深股通净流入")),
            })

        return {
            "data_points": len(records),
            "flows": records,
        }
    except Exception as e:
        return {"error": str(e)}


def screen_stocks(symbols: list[str], filters: dict | None = None) -> dict:
    """
    Screen A-share stocks against financial filters.

    Default filters:
        max_pe: 30        (P/E below 30)
        max_pb: 5         (P/B below 5)
        min_roe: 8        (ROE above 8%)
        max_debt_ratio: 60 (Debt-to-asset ratio below 60%)
    """
    defaults = {
        "max_pe": 30.0,
        "max_pb": 5.0,
        "min_roe": 8.0,
        "max_debt_ratio": 60.0,
    }
    if filters:
        defaults.update(filters)

    passing = []
    failing = []

    for sym in symbols:
        try:
            m = fetch_financial_metrics(sym)
            if "error" in m:
                failing.append({"symbol": sym, "reason": m["error"]})
                continue

            reasons = []
            pe = (m.get("valuation") or {}).get("pe_ttm")
            if pe is not None and pe <= 0:
                reasons.append(f"PE {pe:.1f} 无效（为零或负值，通常表示亏损）")
            elif pe is not None and pe > defaults["max_pe"]:
                reasons.append(f"PE {pe:.1f} > {defaults['max_pe']:.1f}")

            pb = (m.get("valuation") or {}).get("pb")
            if pb is not None and pb > defaults["max_pb"]:
                reasons.append(f"PB {pb:.1f} > {defaults['max_pb']:.1f}")

            roe = (m.get("profitability") or {}).get("roe")
            if roe is not None and roe < defaults["min_roe"]:
                reasons.append(f"ROE {roe:.1f}% < {defaults['min_roe']:.1f}%")

            debt_ratio = (m.get("leverage") or {}).get("debt_to_asset_ratio")
            if debt_ratio is not None and debt_ratio > defaults["max_debt_ratio"]:
                reasons.append(
                    f"Debt ratio {debt_ratio:.1f}% > {defaults['max_debt_ratio']:.1f}%"
                )

            if reasons:
                failing.append({"symbol": sym, "reasons": reasons})
            else:
                passing.append(m)

        except Exception as e:
            failing.append({"symbol": sym, "reason": str(e)})

    return {
        "filters_applied": defaults,
        "total_screened": len(symbols),
        "passed": len(passing),
        "failed": len(failing),
        "results": passing,
        "rejected": failing,
    }


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(
        description="A-Share Stock Data Fetcher (AKShare, no API key)"
    )
    parser.add_argument("symbols", nargs="*", help="A-share stock code(s)")
    parser.add_argument("--metrics", action="store_true",
                        help="Full financial metrics")
    parser.add_argument("--history", action="store_true",
                        help="Price history")
    parser.add_argument("--financials", action="store_true",
                        help="Financial statements")
    parser.add_argument("--insider", action="store_true",
                        help="Insider trading data")
    parser.add_argument("--northbound", action="store_true",
                        help="Northbound capital flow")
    parser.add_argument("--screen", action="store_true",
                        help="Screen against default filters")
    parser.add_argument("--period", default="1y",
                        help="History period (1m,3m,6m,1y,2y,5y,max)")
    args = parser.parse_args()

    try:
        if args.northbound:
            data = fetch_northbound_flow()
        elif not args.symbols:
            error_exit("Please provide stock symbol(s) or use --northbound")
            return
        elif args.screen:
            data = screen_stocks(args.symbols)
        elif args.metrics:
            if len(args.symbols) == 1:
                data = fetch_financial_metrics(args.symbols[0])
            else:
                data = [fetch_financial_metrics(s) for s in args.symbols]
        elif args.history:
            data = fetch_price_history(args.symbols[0], period=args.period)
        elif args.financials:
            data = fetch_financial_statements(args.symbols[0])
        elif args.insider:
            data = fetch_insider_trades(args.symbols[0])
        else:
            data = fetch_basic_info(args.symbols)

        output_json(data)

    except ImportError:
        error_exit("akshare is required. Install: pip install akshare")
    except Exception as e:
        error_exit(f"Error fetching data: {e}")


if __name__ == "__main__":
    main()