path: root/src/FundLab.Api/akshare_collector.py

#!/usr/bin/env python3
"""Small raw-data adapter for the AKShare endpoints used by Fund Lab."""

import argparse
import datetime as dt
import json
import re
import sys
from decimal import Decimal, InvalidOperation

import akshare as ak
import pandas as pd


SCHEMA_VERSION = "fund-lab.akshare.v1"


def collected_at():
    return dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z")


def source_revision():
    return f"akshare-{getattr(ak, '__version__', 'unknown')}/eastmoney"


def is_missing(value):
    try:
        return bool(pd.isna(value))
    except (TypeError, ValueError):
        return False


def text(value):
    if is_missing(value):
        return None
    value = str(value).strip()
    return value or None


def fund_code(value):
    value = text(value)
    if value is None:
        return None
    if value.isdigit() and len(value) < 6:
        value = value.zfill(6)
    return value if re.fullmatch(r"\d{6}", value) else None


def decimal_text(value):
    value = text(value)
    if value is None:
        return None
    try:
        return format(Decimal(value), "f")
    except InvalidOperation:
        return None


def date_text(value):
    if is_missing(value):
        return None
    if isinstance(value, (dt.datetime, dt.date)):
        return value.strftime("%Y-%m-%d")
    parsed = pd.to_datetime(value, errors="coerce")
    if pd.isna(parsed):
        return None
    return parsed.strftime("%Y-%m-%d")


def search(query):
    query = query.strip()
    if not query:
        raise ValueError("search query cannot be empty")

    frame = ak.fund_name_em()
    query_lower = query.casefold()
    rows = []

    for _, row in frame.iterrows():
        code = fund_code(row.get("基金代码"))
        name = text(row.get("基金简称"))
        pinyin = text(row.get("拼音缩写"))
        full_pinyin = text(row.get("拼音全称"))

        if code is None or name is None:
            continue

        searchable = [code, name, pinyin or "", full_pinyin or ""]
        if not any(query_lower in value.casefold() for value in searchable):
            continue

        rows.append(
            {
                "code": code,
                "name": name,
                "fund_type": text(row.get("基金类型")),
                "rank": 0 if code == query else (1 if name == query else 2),
            }
        )

    rows.sort(key=lambda item: (item.pop("rank"), item["code"]))
    return {
        "schema_version": SCHEMA_VERSION,
        "operation": "search",
        "source": "akshare",
        "source_revision": source_revision(),
        "collected_at": collected_at(),
        "instruments": rows[:100],
    }


def nav(code):
    if not re.fullmatch(r"\d{6}", code):
        raise ValueError("fund code must contain exactly six digits")

    frame = ak.fund_open_fund_info_em(
        symbol=code,
        indicator="单位净值走势",
        period="成立来",
    )
    observations = []

    for _, row in frame.iterrows():
        nav_date = date_text(row.get("净值日期"))
        nav_value = decimal_text(row.get("单位净值"))
        if nav_date is None or nav_value is None:
            continue

        observations.append(
            {
                "nav_date": nav_date,
                "published_at": None,
                "nav": nav_value,
                "accumulated_nav": None,
                "daily_return": decimal_text(row.get("日增长率")),
            }
        )

    if not observations:
        raise ValueError(f"AKShare returned no usable NAV observations for {code}")

    return {
        "schema_version": SCHEMA_VERSION,
        "operation": "nav",
        "source": "akshare",
        "source_revision": source_revision(),
        "collected_at": collected_at(),
        "instrument": {"code": code},
        "observations": observations,
    }


def bond_maturity_date(code):
    # Maturity is a best-effort enrichment: the quote endpoint has no expiry
    # date, so a failure here only means maturity_date stays null.
    try:
        frame = ak.bond_zh_cov_info(symbol=code, indicator="基本信息")
        if frame is None or frame.empty:
            return None
        return date_text(frame.iloc[0].get("EXPIRE_DATE"))
    except Exception:
        return None


def bond_quote(code):
    code = text(code)
    if code is None or not re.fullmatch(r"\d{6}", code):
        raise ValueError("bond code must contain exactly six digits")

    frame = ak.bond_zh_hs_cov_spot()
    if frame is None or frame.empty:
        raise ValueError(f"no valid quote for {code}")

    prefix_rank = {"sh": 0, "sz": 1, "bj": 2}
    candidates = []

    for _, row in frame.iterrows():
        symbol = text(row.get("symbol")) or ""
        row_code = text(row.get("code")) or ""
        if row_code != code and not symbol.endswith(code):
            continue

        price = decimal_text(row.get("trade"))
        if price is None or Decimal(price) <= 0:
            continue

        prefix = symbol[:2].lower()
        candidates.append((prefix_rank.get(prefix, 3), symbol, text(row.get("name")), price))

    if not candidates:
        raise ValueError(f"no valid quote for {code}")

    _, _, name, price = sorted(candidates, key=lambda item: (item[0], item[1]))[0]

    return {
        "schema_version": SCHEMA_VERSION,
        "operation": "bond-quote",
        "source": "akshare",
        "source_revision": source_revision(),
        "collected_at": collected_at(),
        "instrument": {"code": code},
        "quote": {
            "name": name,
            "price": price,
            "maturity_date": bond_maturity_date(code),
        },
    }


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--operation", choices=("search", "nav", "bond-quote"), required=True)
    parser.add_argument("--query")
    parser.add_argument("--code")
    args = parser.parse_args()

    try:
        if args.operation == "search":
            payload = search(args.query)
        elif args.operation == "nav":
            payload = nav(args.code)
        else:
            payload = bond_quote(args.code)

        json.dump(payload, sys.stdout, ensure_ascii=False, separators=(",", ":"))
        sys.stdout.write("\n")
        return 0
    except Exception as error:
        print(f"AKShare collector failed: {error}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())