""" 东方财富财务摘要 API (FastAPI 封装) GET /fin/summary?stocks=603233.SH:10500736,600519.SH:xxxxxx GET /fin/summary?stocks=603233.SH (只传代码, 自动查 ORG_CODE, 不稳) 返回 JSON: {success, count, columns, data, errors} """ from __future__ import annotations from typing import Optional from fastapi import FastAPI, Query, HTTPException from fastapi.responses import JSONResponse import pandas as pd # 复用 fin_summary.py 里的核心逻辑 from utils.fin_summary import get_fin_summary, _fetch_org_codes from utils.fin_quarterly import get_fin_quarterly_summary from utils.fin_main_business import get_main_business from utils.fin_balance import get_balance from utils.fin_income import get_income from utils.fin_cashflow import get_cashflow from utils.fin_profitability import get_profitability from utils.fin_capital_structure import get_capital_structure from utils.fin_operate import get_operate_ability from utils.fin_growth import get_growth from utils.fin_dupont import get_dupont from utils.fin_dividend import get_dividend_detail, get_dividend_statics from utils.fin_free_holders import get_free_holders from utils.fin_holders import get_holders from utils.fin_holder_num import get_holder_num from utils.fin_mda import get_mda from utils.fin_equity import get_equity from utils.fin_company import get_company_info, get_industry from utils.fin_belong_index import get_index_classif, get_belong_index, get_index_membership from utils.fin_major_events import ( get_acquisitions, get_equity_incentive, get_litigation, get_related_transaction, get_violation, get_guarantee, ) app = FastAPI( title="EMWeb API", description="东方财富 CHOICE API 数据抓取服务", version="0.1.0", ) def _parse_stocks(raw: str) -> tuple[list[str], Optional[dict[str, str]]]: """ 解析 ?stocks= 参数. "603233.SH:10500736,600519.SH:xxxxxx" -> (codes, {code: org_code}) "603233.SH,600519.SH" -> (codes, None) -> 自动查 "603233.SH:10500736" -> (codes, {code: org_code}) 单只 """ codes: list[str] = [] org_map: dict[str, str] = {} has_any_org = False for part in raw.split(","): part = part.strip() if not part: continue if ":" in part: code, org = part.split(":", 1) code = code.strip() org = org.strip() codes.append(code) if org: org_map[code] = org has_any_org = True else: codes.append(part.strip()) if has_any_org and len(org_map) != len(codes): # 部分给了 ORG_CODE 部分没给 -> 退化为全部自动查 return codes, None return codes, (org_map if has_any_org else None) def _df_to_json_payload(df: pd.DataFrame) -> dict: """DataFrame -> JSON-safe dict.""" import math import re safe = df.copy() # 处理 NaN / NaT / Timestamp / inf -> None for col in safe.columns: if pd.api.types.is_datetime64_any_dtype(safe[col]): safe[col] = safe[col].astype(str).replace("NaT", None) safe = safe.where(pd.notnull(safe), None) # 二次清理: inf / -inf / 极大值 records = safe.to_dict(orient="records") cleaned = [] for row in records: new_row = {} for k, v in row.items(): if v is None: new_row[k] = None elif isinstance(v, float): if math.isnan(v) or math.isinf(v): new_row[k] = None else: new_row[k] = v elif isinstance(v, str): # 去除 ASCII 控制字符 (除 \t \n \r 外), 防止 JSON 解析失败 # 长文本字段 (如 ORG_PROFILE) 可能含 \f \v 等非标准换行 v = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', v) new_row[k] = v else: new_row[k] = v cleaned.append(new_row) return { "success": True, "count": int(len(cleaned)), "columns": list(safe.columns), "data": cleaned, } def _empty_response(message: str, codes: list[str], **extra) -> JSONResponse: """无数据时的统一 404 返回.""" content = { "success": False, "count": 0, "message": message, "codes_requested": codes, "data": [], } content.update(extra) return JSONResponse(status_code=404, content=content) @app.get("/") def root(): return { "name": "EMWeb API", "version": "0.1.0", "endpoints": { "GET /fin/summary": "拉取财务摘要(报告期)", "GET /fin/quarterly-summary": "拉取季度财务摘要", "GET /fin/main-business": "拉取主营构成(按产品/按地区)", "GET /fin/balance": "拉取资产负债表", "GET /fin/income": "拉取利润表", "GET /fin/cashflow": "拉取现金流量表", "GET /fin/profitability": "拉取盈利能力与收益质量", "GET /fin/capital-structure": "拉取资本结构与偿债能力", "GET /fin/operate-ability": "拉取营运能力", "GET /fin/growth": "拉取成长能力", "GET /fin/dupont": "拉取杜邦分析", "GET /fin/dividend-detail": "拉取分红明细", "GET /fin/dividend-statics": "拉取分红统计", "GET /fin/free-holders": "拉取十大流通股东", "GET /fin/holders": "拉取十大股东明细", "GET /fin/holder-num": "拉取股东户数", "GET /fin/mda": "拉取管理层讨论与分析(年报/中报)", "GET /fin/equity": "拉取股本结构", "GET /fin/company-info": "拉取公司介绍", "GET /fin/industry": "拉取所属行业", "GET /fin/index-classif": "拉取所属指数分类", "GET /fin/belong-index": "拉取纳入指数统计", "GET /fin/index-membership": "拉取指数成分权重", "GET /fin/acquisitions": "拉取并购事件", "GET /fin/equity-incentive": "拉取股权激励", "GET /fin/litigation": "拉取诉讼仲裁", "GET /fin/related-transaction": "拉取关联交易", "GET /fin/violation": "拉取违规处罚", "GET /fin/guarantee": "拉取对外担保", "GET /health": "健康检查", }, "usage": { "fin_summary": "/fin/summary?stocks=603233.SH:10500736&years=10", "fin_quarterly_summary": "/fin/quarterly-summary?stocks=603233.SH&years=3", "fin_main_business": "/fin/main-business?stocks=603233.SH&classify_type=产品&years=3", "fin_balance": "/fin/balance?stocks=603233.SH&years=3", "fin_dividend_detail": "/fin/dividend-detail?stocks=603233.SH&years=5", "fin_mda": "/fin/mda?stocks=603233.SH&years=5", "fin_company_info": "/fin/company-info?stocks=603233.SH", }, } @app.get("/health") def health(): return {"status": "ok"} @app.get("/fin/summary") def fin_summary( stocks: str = Query( ..., description=( "股票列表, 逗号分隔. 支持仅代码自动查询；其中 '603233.SH' " "会自动解析为 ORG_CODE '10500736'。也可使用代码:ORG_CODE，" "如 '603233.SH:10500736,600519.SH:xxxxxx'" ), examples=["603233.SH", "603233.SH:10500736"], ), years: int = Query(10, ge=1, le=20, description="拉取多少年的历史"), date_type_codes: str = Query( "1,5,6", description="报告期代码, 逗号分隔. 1=一季报, 5=中报, 6=年报", ), is_newest: bool = Query(True, description="是否只要最新合并报表"), ): """拉取多只股票的财务摘要数据, 返回 JSON.""" codes, org_map = _parse_stocks(stocks) if not codes: raise HTTPException(400, detail="stocks 参数不能为空") # 自动查 ORG_CODE if org_map is None: try: org_map = _fetch_org_codes(codes) except Exception as e: # noqa: BLE001 raise HTTPException( 400, detail=( f"自动查询失败: {e}. 请改用 '代码:ORG_CODE' 格式, " f"ORG_CODE 可在 Reqable 抓包切换股票时拿到." ), ) df = get_fin_summary( org_map, years=years, date_type_codes=date_type_codes, is_newest=is_newest, verbose=False, ) if df.empty: return JSONResponse( status_code=404, content={ "success": False, "count": 0, "message": "无数据, 请检查 ORG_CODE 或放宽筛选条件", "codes_requested": codes, "data": [], }, ) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years payload["date_type_codes"] = date_type_codes return payload @app.get("/fin/quarterly-summary") def fin_quarterly_summary( stocks: str = Query( ..., description="股票代码，逗号分隔，例如 '603233.SH,600519.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"], ), years: int = Query(3, ge=1, le=20, description="未传 report_dates 时拉取最近多少年的季度数据"), report_dates: Optional[str] = Query( None, description=( "可选，逗号分隔的 YYYY-MM-DD 报告期。未传时取最近 years 年季度数据；" "传入时精确筛选报告期，并按股票代码和报告期自动去重。" ), examples=["2025-03-31,2024-12-31"], ), ): """拉取季度财务摘要；指定报告期时执行精确报告期筛选。""" codes = [code.strip() for code in stocks.split(",") if code.strip()] if not codes: raise HTTPException(400, detail="stocks 参数不能为空") dates = None if report_dates is not None: dates = [date.strip() for date in report_dates.split(",") if date.strip()] if not dates: raise HTTPException(400, detail="report_dates 参数不能为空") df = get_fin_quarterly_summary(codes, years=years, report_dates=dates) if df.empty: return JSONResponse( status_code=404, content={ "success": False, "count": 0, "message": "无季度财务摘要数据, 请检查股票代码或筛选条件", "codes_requested": codes, "years": years, "data": [], }, ) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload @app.get("/fin/main-business") def fin_main_business( stocks: str = Query( ..., description="股票代码，逗号分隔，例如 '603233.SH,600519.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"], ), classify_type: str = Query( "产品", description="分类方式: 产品 或 地区", examples=["产品", "地区"], ), years: int = Query(3, ge=1, le=20, description="拉取最近多少年的主营构成数据"), ): """拉取主营构成(按产品或按地区)。""" codes = [code.strip() for code in stocks.split(",") if code.strip()] if not codes: raise HTTPException(400, detail="stocks 参数不能为空") df = get_main_business(codes, classify_type=classify_type, years=years) if df.empty: return JSONResponse( status_code=404, content={ "success": False, "count": 0, "message": "无主营构成数据, 请检查股票代码或筛选条件", "codes_requested": codes, "classify_type": classify_type, "data": [], }, ) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["classify_type"] = classify_type payload["years"] = years return payload @app.get("/fin/balance") def fin_balance( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), years: int = Query(3, ge=1, le=20, description="拉取最近多少年的资产负债表数据"), ): """拉取资产负债表。""" codes = [code.strip() for code in stocks.split(",") if code.strip()] if not codes: raise HTTPException(400, detail="stocks 参数不能为空") df = get_balance(codes, years=years) if df.empty: return _empty_response("无资产负债表数据, 请检查股票代码或筛选条件", codes, years=years) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload @app.get("/fin/income") def fin_income( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), years: int = Query(3, ge=1, le=20, description="拉取最近多少年的利润表数据"), ): """拉取利润表。""" codes = [code.strip() for code in stocks.split(",") if code.strip()] if not codes: raise HTTPException(400, detail="stocks 参数不能为空") df = get_income(codes, years=years) if df.empty: return _empty_response("无利润表数据, 请检查股票代码或筛选条件", codes, years=years) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload @app.get("/fin/cashflow") def fin_cashflow( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), years: int = Query(3, ge=1, le=20, description="拉取最近多少年的现金流量表数据"), ): """拉取现金流量表。""" codes = [code.strip() for code in stocks.split(",") if code.strip()] if not codes: raise HTTPException(400, detail="stocks 参数不能为空") df = get_cashflow(codes, years=years) if df.empty: return _empty_response("无现金流量表数据, 请检查股票代码或筛选条件", codes, years=years) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload @app.get("/fin/profitability") def fin_profitability( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), years: int = Query(3, ge=1, le=20, description="拉取最近多少年的盈利能力与收益质量数据"), ): """拉取盈利能力与收益质量。""" codes = [code.strip() for code in stocks.split(",") if code.strip()] if not codes: raise HTTPException(400, detail="stocks 参数不能为空") df = get_profitability(codes, years=years) if df.empty: return _empty_response("无盈利能力与收益质量数据, 请检查股票代码或筛选条件", codes, years=years) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload @app.get("/fin/capital-structure") def fin_capital_structure( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), years: int = Query(3, ge=1, le=20, description="拉取最近多少年的资本结构与偿债能力数据"), ): """拉取资本结构与偿债能力。""" codes = [code.strip() for code in stocks.split(",") if code.strip()] if not codes: raise HTTPException(400, detail="stocks 参数不能为空") df = get_capital_structure(codes, years=years) if df.empty: return _empty_response("无资本结构与偿债能力数据, 请检查股票代码或筛选条件", codes, years=years) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload @app.get("/fin/operate-ability") def fin_operate_ability( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), years: int = Query(3, ge=1, le=20, description="拉取最近多少年的营运能力数据"), ): """拉取营运能力。""" codes = [code.strip() for code in stocks.split(",") if code.strip()] if not codes: raise HTTPException(400, detail="stocks 参数不能为空") df = get_operate_ability(codes, years=years) if df.empty: return _empty_response("无营运能力数据, 请检查股票代码或筛选条件", codes, years=years) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload @app.get("/fin/growth") def fin_growth( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), years: int = Query(3, ge=1, le=20, description="拉取最近多少年的成长能力数据"), ): """拉取成长能力。""" codes = [code.strip() for code in stocks.split(",") if code.strip()] if not codes: raise HTTPException(400, detail="stocks 参数不能为空") df = get_growth(codes, years=years) if df.empty: return _empty_response("无成长能力数据, 请检查股票代码或筛选条件", codes, years=years) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload @app.get("/fin/dupont") def fin_dupont( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), years: int = Query(3, ge=1, le=20, description="拉取最近多少年的杜邦分析数据"), ): """拉取杜邦分析。""" codes = [code.strip() for code in stocks.split(",") if code.strip()] if not codes: raise HTTPException(400, detail="stocks 参数不能为空") df = get_dupont(codes, years=years) if df.empty: return _empty_response("无杜邦分析数据, 请检查股票代码或筛选条件", codes, years=years) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload @app.get("/fin/dividend-detail") def fin_dividend_detail( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), years: int = Query(5, ge=1, le=20, description="拉取最近多少年的分红明细"), ): """拉取分红明细。""" codes = [code.strip() for code in stocks.split(",") if code.strip()] if not codes: raise HTTPException(400, detail="stocks 参数不能为空") df = get_dividend_detail(codes, years=years) if df.empty: return _empty_response("无分红明细数据, 请检查股票代码或筛选条件", codes, years=years) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload @app.get("/fin/dividend-statics") def fin_dividend_statics( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), years: int = Query(5, ge=1, le=20, description="拉取最近多少年的分红统计"), ): """拉取分红统计。""" codes = [code.strip() for code in stocks.split(",") if code.strip()] if not codes: raise HTTPException(400, detail="stocks 参数不能为空") df = get_dividend_statics(codes, years=years) if df.empty: return _empty_response("无分红统计数据, 请检查股票代码或筛选条件", codes, years=years) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload @app.get("/fin/free-holders") def fin_free_holders( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), years: int = Query(3, ge=1, le=20, description="拉取最近多少年的十大流通股东"), ): """拉取十大流通股东。""" codes = [code.strip() for code in stocks.split(",") if code.strip()] if not codes: raise HTTPException(400, detail="stocks 参数不能为空") df = get_free_holders(codes, years=years) if df.empty: return _empty_response("无十大流通股东数据, 请检查股票代码或筛选条件", codes, years=years) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload @app.get("/fin/holders") def fin_holders( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), years: int = Query(3, ge=1, le=20, description="拉取最近多少年的十大股东明细"), ): """拉取十大股东明细。""" codes = [code.strip() for code in stocks.split(",") if code.strip()] if not codes: raise HTTPException(400, detail="stocks 参数不能为空") df = get_holders(codes, years=years) if df.empty: return _empty_response("无十大股东明细数据, 请检查股票代码或筛选条件", codes, years=years) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload @app.get("/fin/holder-num") def fin_holder_num( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), years: int = Query(3, ge=1, le=20, description="拉取最近多少年的股东户数"), ): """拉取股东户数。""" codes = [code.strip() for code in stocks.split(",") if code.strip()] if not codes: raise HTTPException(400, detail="stocks 参数不能为空") df = get_holder_num(codes, years=years) if df.empty: return _empty_response("无股东户数数据, 请检查股票代码或筛选条件", codes, years=years) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload def _split_codes(stocks: str) -> list[str]: """解析逗号分隔的股票代码参数.""" codes = [code.strip() for code in stocks.split(",") if code.strip()] if not codes: raise HTTPException(400, detail="stocks 参数不能为空") return codes @app.get("/fin/mda") def fin_mda( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), years: int = Query(5, ge=1, le=20, description="拉取最近多少年的管理层讨论与分析"), ): """拉取管理层讨论与分析(年报/中报)。""" codes = _split_codes(stocks) df = get_mda(codes, years=years) if df.empty: return _empty_response("无管理层讨论与分析数据, 请检查股票代码或筛选条件", codes, years=years) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload @app.get("/fin/equity") def fin_equity( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), ): """拉取股本结构 (全量历史)。""" codes = _split_codes(stocks) df = get_equity(codes) if df.empty: return _empty_response("无股本结构数据, 请检查股票代码或筛选条件", codes) payload = _df_to_json_payload(df) payload["codes_requested"] = codes return payload @app.get("/fin/company-info") def fin_company_info( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), ): """拉取公司介绍 (基本资料)。""" codes = _split_codes(stocks) df = get_company_info(codes) if df.empty: return _empty_response("无公司介绍数据, 请检查股票代码或筛选条件", codes) payload = _df_to_json_payload(df) payload["codes_requested"] = codes return payload @app.get("/fin/industry") def fin_industry( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), ): """拉取所属行业 (当前生效分类)。""" codes = _split_codes(stocks) df = get_industry(codes) if df.empty: return _empty_response("无所属行业数据, 请检查股票代码或筛选条件", codes) payload = _df_to_json_payload(df) payload["codes_requested"] = codes return payload @app.get("/fin/index-classif") def fin_index_classif( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), ): """拉取所属指数分类。""" codes = _split_codes(stocks) df = get_index_classif(codes) if df.empty: return _empty_response("无所属指数分类数据, 请检查股票代码或筛选条件", codes) payload = _df_to_json_payload(df) payload["codes_requested"] = codes return payload @app.get("/fin/belong-index") def fin_belong_index( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), ): """拉取纳入指数统计。""" codes = _split_codes(stocks) df = get_belong_index(codes) if df.empty: return _empty_response("无纳入指数统计数据, 请检查股票代码或筛选条件", codes) payload = _df_to_json_payload(df) payload["codes_requested"] = codes return payload @app.get("/fin/index-membership") def fin_index_membership( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), ): """拉取指数成分权重。""" codes = _split_codes(stocks) df = get_index_membership(codes) if df.empty: return _empty_response("无指数成分权重数据, 请检查股票代码或筛选条件", codes) payload = _df_to_json_payload(df) payload["codes_requested"] = codes return payload @app.get("/fin/acquisitions") def fin_acquisitions( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), years: int = Query(3, ge=1, le=20, description="拉取最近多少年的并购事件"), ): """拉取并购事件。""" codes = _split_codes(stocks) df = get_acquisitions(codes, years=years) if df.empty: return _empty_response("无并购事件数据, 请检查股票代码或筛选条件", codes, years=years) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload @app.get("/fin/equity-incentive") def fin_equity_incentive( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), years: int = Query(3, ge=1, le=20, description="拉取最近多少年的股权激励"), ): """拉取股权激励方案。""" codes = _split_codes(stocks) df = get_equity_incentive(codes, years=years) if df.empty: return _empty_response("无股权激励数据, 请检查股票代码或筛选条件", codes, years=years) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload @app.get("/fin/litigation") def fin_litigation( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), years: int = Query(3, ge=1, le=20, description="拉取最近多少年的诉讼仲裁"), ): """拉取诉讼仲裁信息。""" codes = _split_codes(stocks) df = get_litigation(codes, years=years) if df.empty: return _empty_response("无诉讼仲裁数据, 请检查股票代码或筛选条件", codes, years=years) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload @app.get("/fin/related-transaction") def fin_related_transaction( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), years: int = Query(3, ge=1, le=20, description="拉取最近多少年的关联交易"), ): """拉取关联交易。""" codes = _split_codes(stocks) df = get_related_transaction(codes, years=years) if df.empty: return _empty_response("无关联交易数据, 请检查股票代码或筛选条件", codes, years=years) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload @app.get("/fin/violation") def fin_violation( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), years: int = Query(3, ge=1, le=20, description="拉取最近多少年的违规处罚"), ): """拉取违规处罚记录。""" codes = _split_codes(stocks) df = get_violation(codes, years=years) if df.empty: return _empty_response("无违规处罚数据, 请检查股票代码或筛选条件", codes, years=years) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload @app.get("/fin/guarantee") def fin_guarantee( stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]), years: int = Query(3, ge=1, le=20, description="拉取最近多少年的对外担保"), ): """拉取对外担保记录。""" codes = _split_codes(stocks) df = get_guarantee(codes, years=years) if df.empty: return _empty_response("无对外担保数据, 请检查股票代码或筛选条件", codes, years=years) payload = _df_to_json_payload(df) payload["codes_requested"] = codes payload["years"] = years return payload if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8765, log_level="info")