from __future__ import annotations from collections import Counter from dataclasses import dataclass from datetime import date from pathlib import Path import tomllib import duckdb import numpy as np import pandas as pd from .config import Settings FUNDAMENTAL_MODEL_VERSION = "0.3.0" TTM_SOURCE_FIELDS = { "operating_revenue_ytd": "operating_revenue_ttm_calc", "parent_net_profit_ytd": "parent_net_profit_ttm_calc", "deducted_net_profit_ytd": "deducted_net_profit_ttm_calc", "operating_cash_flow_ytd": "operating_cash_flow_ttm_calc", "capital_expenditure_cash_ytd": "capital_expenditure_ttm_calc", } @dataclass(frozen=True) class FundamentalBuildResult: rows: int scored: int hard_failed: int output_path: Path def _safe_div(numerator: pd.Series, denominator: pd.Series) -> pd.Series: denominator = pd.to_numeric(denominator, errors="coerce").replace(0, np.nan) result = pd.to_numeric(numerator, errors="coerce") / denominator return result.replace([np.inf, -np.inf], np.nan) def _cagr(current: pd.Series, previous: pd.Series, years: int) -> pd.Series: current = pd.to_numeric(current, errors="coerce") previous = pd.to_numeric(previous, errors="coerce") valid = current.gt(0) & previous.gt(0) result = pd.Series(np.nan, index=current.index, dtype=float) result.loc[valid] = (current.loc[valid] / previous.loc[valid]).pow(1.0 / years) - 1.0 return result * 100.0 def _industry_percentile(frame: pd.DataFrame, column: str, higher_is_better: bool = True) -> pd.Series: values = pd.to_numeric(frame[column], errors="coerce").replace([np.inf, -np.inf], np.nan) ranked = values.groupby(frame["industry_l2_code"]).rank(pct=True, ascending=higher_is_better) return ranked def _component(frame: pd.DataFrame, specs: list[tuple[str, bool]], weight: float) -> tuple[pd.Series, pd.Series]: pieces = [] for column, higher_is_better in specs: pieces.append(_industry_percentile(frame, column, higher_is_better).rename(column)) matrix = pd.concat(pieces, axis=1) coverage = matrix.notna().mean(axis=1) score = matrix.mean(axis=1, skipna=True) * weight return score.where(coverage.gt(0)), coverage def _load_rules(settings: Settings) -> dict[str, object]: path = settings.project_root / "config" / "fundamental_rules.toml" return tomllib.loads(path.read_text(encoding="utf-8")) def _add_calculated_ttm(finance: pd.DataFrame) -> pd.DataFrame: """Build rolling-four-quarter values from cumulative statement fields. Some TDX convenience TTM fields use mixed units. The cumulative income and cash-flow statement fields have a consistent currency unit, so we first recover single-quarter values and then sum four consecutive quarters. """ result = finance.sort_values(["source_code", "report_period", "available_trade_date"]).copy() result = result.drop_duplicates(["source_code", "report_period"], keep="last") result["fiscal_year"] = result["report_period"].dt.year result["fiscal_quarter"] = result["report_period"].dt.quarter grouped = result.groupby(["source_code", "fiscal_year"], sort=False) previous_quarter = grouped["fiscal_quarter"].shift(1) for ytd_column, ttm_column in TTM_SOURCE_FIELDS.items(): values = pd.to_numeric(result[ytd_column], errors="coerce") previous_ytd = grouped[ytd_column].shift(1) is_q1 = result["fiscal_quarter"].eq(1) is_consecutive = previous_quarter.eq(result["fiscal_quarter"] - 1) single_quarter = values.where(is_q1, values - previous_ytd) single_quarter = single_quarter.where(is_q1 | is_consecutive) single_column = ttm_column.replace("_ttm_calc", "_single_quarter_calc") result[single_column] = single_quarter result[ttm_column] = ( result.groupby("source_code", sort=False)[single_column] .rolling(4, min_periods=4) .sum() .reset_index(level=0, drop=True) ) return result.drop(columns=["fiscal_year", "fiscal_quarter"]) def _select_asof_finance(finance: pd.DataFrame, asof_date: pd.Timestamp) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: usable = finance.loc[finance["available_trade_date"].le(asof_date)].copy() usable = usable.sort_values(["source_code", "available_trade_date", "report_period"]) current = usable.drop_duplicates("source_code", keep="last").copy() one_year = current[["source_code", "report_period"]].copy() one_year["report_period"] = one_year["report_period"] - pd.DateOffset(years=1) three_year = current[["source_code", "report_period"]].copy() three_year["report_period"] = three_year["report_period"] - pd.DateOffset(years=3) lookup_columns = [ "source_code", "report_period", "accounts_receivable", "inventory", "total_assets", "parent_equity", "operating_revenue_ytd", "parent_net_profit_ytd", "operating_revenue_ttm_calc", "parent_net_profit_ttm_calc", "deducted_net_profit_ttm_calc", "operating_cash_flow_ttm_calc", "capital_expenditure_ttm_calc", ] history = usable[lookup_columns].drop_duplicates(["source_code", "report_period"], keep="last") previous_1y = one_year.merge(history, on=["source_code", "report_period"], how="left") previous_3y = three_year.merge(history, on=["source_code", "report_period"], how="left") return current.reset_index(drop=True), previous_1y.reset_index(drop=True), previous_3y.reset_index(drop=True) def _add_metrics(frame: pd.DataFrame, previous_1y: pd.DataFrame, previous_3y: pd.DataFrame) -> pd.DataFrame: result = frame.copy().reset_index(drop=True) p1 = previous_1y.reset_index(drop=True) p3 = previous_3y.reset_index(drop=True) short_debt = result["short_term_borrowings"] + result["noncurrent_liabilities_due_1y"] result["debt_ratio"] = _safe_div(result["total_liabilities"], result["total_assets"]) * 100 result["current_ratio"] = _safe_div(result["current_assets"], result["current_liabilities"]) result["quick_ratio"] = _safe_div(result["current_assets"] - result["inventory"], result["current_liabilities"]) result["cash_short_debt_ratio"] = _safe_div(result["money_funds"], short_debt) result["goodwill_equity_ratio"] = _safe_div(result["goodwill"], result["parent_equity"]) result["cfo_to_parent_profit"] = _safe_div(result["operating_cash_flow_ttm_calc"], result["parent_net_profit_ttm_calc"]) result["cfo_margin"] = _safe_div(result["operating_cash_flow_ttm_calc"], result["operating_revenue_ttm_calc"]) * 100 result["net_margin"] = _safe_div(result["parent_net_profit_ttm_calc"], result["operating_revenue_ttm_calc"]) * 100 result["roa_proxy"] = _safe_div(result["parent_net_profit_ttm_calc"], result["total_assets"]) * 100 result["research_development_ratio"] = _safe_div( result["research_development_expense_ytd"], result["operating_revenue_ytd"] ) * 100 result["receivables_yoy"] = (_safe_div(result["accounts_receivable"], p1["accounts_receivable"]) - 1) * 100 result["inventory_yoy"] = (_safe_div(result["inventory"], p1["inventory"]) - 1) * 100 result["parent_equity_yoy"] = (_safe_div(result["parent_equity"], p1["parent_equity"]) - 1) * 100 result["equity_assets_ratio"] = _safe_div(result["parent_equity"], result["total_assets"]) * 100 result["deducted_profit_ratio"] = _safe_div(result["deducted_net_profit_ttm_calc"], result["parent_net_profit_ttm_calc"]) result["impairment_revenue_ratio"] = _safe_div(result["impairment_loss_ytd"], result["operating_revenue_ytd"]).abs() result["parent_net_profit_ttm_previous_1y"] = p1["parent_net_profit_ttm_calc"] result["cfo_ttm_yoy"] = (_safe_div(result["operating_cash_flow_ttm_calc"], p1["operating_cash_flow_ttm_calc"]) - 1) * 100 result["revenue_cagr_3y"] = _cagr(result["operating_revenue_ttm_calc"], p3["operating_revenue_ttm_calc"], 3) result["parent_profit_cagr_3y"] = _cagr(result["parent_net_profit_ttm_calc"], p3["parent_net_profit_ttm_calc"], 3) result["market_cap"] = result["latest_close"] * result["total_shares"] result["pe_ttm"] = _safe_div(result["market_cap"], result["parent_net_profit_ttm_calc"]).where(result["parent_net_profit_ttm_calc"].gt(0)) result["pb"] = _safe_div(result["market_cap"], result["parent_equity"]).where(result["parent_equity"].gt(0)) result["ps_ttm"] = _safe_div(result["market_cap"], result["operating_revenue_ttm_calc"]).where(result["operating_revenue_ttm_calc"].gt(0)) return result def _risk_flags(frame: pd.DataFrame, rules: dict[str, float]) -> pd.DataFrame: rows: list[dict[str, object]] = [] def add(mask: pd.Series, code: str, severity: str, description: str, value_column: str) -> None: for row in frame.loc[mask.fillna(False), ["security_id", "source_code", value_column]].itertuples(index=False): rows.append({ "security_id": row.security_id, "source_code": row.source_code, "rule_code": code, "severity": severity, "description": description, "metric_value": getattr(row, value_column), "metric_name": value_column, }) non_financial = ~frame["is_financial_sector"] add(frame["parent_equity"].le(0), "NEGATIVE_EQUITY", "hard", "归母净资产不为正", "parent_equity") add(frame["total_assets"].le(0), "INVALID_TOTAL_ASSETS", "hard", "总资产不为正", "total_assets") add(non_financial & frame["debt_ratio"].ge(rules["debt_ratio_hard"]), "DEBT_RATIO_GE_100", "hard", "资产负债率达到或超过100%", "debt_ratio") add(non_financial & frame["goodwill_equity_ratio"].ge(rules["goodwill_equity_hard"]), "GOODWILL_EQUITY_EXTREME", "hard", "商誉超过归母净资产80%", "goodwill_equity_ratio") add(non_financial & frame["current_ratio"].lt(rules["current_ratio_warning"]) & frame["operating_cash_flow_ttm_calc"].lt(0), "LIQUIDITY_AND_CFO_WEAK", "warning", "流动比率偏低且近一年经营现金流为负", "current_ratio") add(non_financial & frame["cash_short_debt_ratio"].lt(rules["cash_short_debt_warning"]) & frame["operating_cash_flow_ttm_calc"].lt(0), "CASH_SHORT_DEBT_WEAK", "warning", "货币资金覆盖短债不足且经营现金流为负", "cash_short_debt_ratio") add(non_financial & frame["parent_net_profit_ttm_calc"].gt(0) & frame["cfo_to_parent_profit"].lt(rules["cash_profit_warning"]), "CASH_PROFIT_MISMATCH", "warning", "经营现金流与归母利润匹配偏弱", "cfo_to_parent_profit") add(non_financial & frame["receivables_yoy"].sub(frame["operating_revenue_yoy_reported"]).gt(rules["receivable_growth_gap_warning_pct"]), "RECEIVABLES_GROWTH_GAP", "warning", "应收账款增速明显高于营收增速", "receivables_yoy") add(non_financial & frame["inventory_yoy"].sub(frame["operating_revenue_yoy_reported"]).gt(rules["inventory_growth_gap_warning_pct"]), "INVENTORY_GROWTH_GAP", "warning", "存货增速明显高于营收增速", "inventory_yoy") add(non_financial & frame["goodwill_equity_ratio"].between(rules["goodwill_equity_warning"], rules["goodwill_equity_hard"], inclusive="left"), "GOODWILL_EQUITY_HIGH", "warning", "商誉占归母净资产比例较高", "goodwill_equity_ratio") add(non_financial & frame["parent_net_profit_ttm_calc"].lt(0) & frame["operating_revenue_yoy_reported"].lt(0), "LOSS_AND_REVENUE_DECLINE", "warning", "近一年亏损且营收同比下降", "parent_net_profit_ttm_calc") return pd.DataFrame(rows, columns=["security_id", "source_code", "rule_code", "severity", "description", "metric_value", "metric_name"]) def _add_financial_risk_flags(frame: pd.DataFrame, flags: pd.DataFrame, rules: dict[str, float]) -> pd.DataFrame: rows: list[dict[str, object]] = [] financial = frame["is_financial_sector"] def add(mask: pd.Series, code: str, description: str, value_column: str) -> None: for row in frame.loc[mask.fillna(False), ["security_id", "source_code", value_column]].itertuples(index=False): rows.append({ "security_id": row.security_id, "source_code": row.source_code, "rule_code": code, "severity": "warning", "description": description, "metric_value": getattr(row, value_column), "metric_name": value_column, }) add(financial & frame["parent_net_profit_ttm_calc"].lt(0), "FINANCIAL_TTM_LOSS", "金融企业近四季度归母净利润亏损", "parent_net_profit_ttm_calc") add(financial & frame["parent_equity_yoy"].lt(rules["equity_decline_warning_pct"]), "FINANCIAL_EQUITY_DECLINE", "归母净资产同比下降超过20%", "parent_equity_yoy") add(financial & frame["impairment_revenue_ratio"].gt(rules["impairment_revenue_warning"]), "FINANCIAL_IMPAIRMENT_HIGH", "减值损失占营收比例较高", "impairment_revenue_ratio") add(financial & frame["parent_net_profit_ttm_calc"].gt(0) & frame["deducted_profit_ratio"].lt(rules["deducted_profit_ratio_warning"]), "FINANCIAL_DEDUCTED_PROFIT_WEAK", "扣非净利润对归母净利润的覆盖偏低", "deducted_profit_ratio") return pd.concat([flags, pd.DataFrame(rows, columns=flags.columns)], ignore_index=True) def _add_external_risk_flags(frame: pd.DataFrame, flags: pd.DataFrame, settings: Settings, rules: dict[str, float]) -> pd.DataFrame: external_path = settings.processed_dir / "external_governance_risk_flags.csv" pledge_path = settings.processed_dir / "external_pledge_latest.parquet" additions: list[pd.DataFrame] = [] if external_path.exists(): external = pd.read_csv(external_path) if not external.empty: external["event_date"] = pd.to_datetime(external["event_date"], errors="coerce") cutoff = pd.Timestamp.today().normalize() - pd.Timedelta(days=365) external = external.loc[external["event_date"].ge(cutoff)].copy() external = external.sort_values("event_date").drop_duplicates( ["source_code", "rule_code"], keep="last" ) current_st = frame.set_index("source_code")["is_st_current"] delisting = external["rule_code"].eq("DELISTING_RISK") external = external.loc[~delisting | external["source_code"].map(current_st).fillna(False)].copy() external["metric_value"] = np.nan external["metric_name"] = "announcement_title_match" additions.append(external) if pledge_path.exists(): pledge = pd.read_parquet(pledge_path) pledge = frame[["security_id", "source_code"]].merge( pledge[["source_code", "pledge_ratio_pct"]], on="source_code", how="inner" ) rows: list[dict[str, object]] = [] for row in pledge.itertuples(index=False): if row.pledge_ratio_pct >= rules["pledge_ratio_hard_pct"]: severity, code, description = "hard", "PLEDGE_RATIO_EXTREME", "股权质押比例达到70%或以上" elif row.pledge_ratio_pct >= rules["pledge_ratio_warning_pct"]: severity, code, description = "warning", "PLEDGE_RATIO_HIGH", "股权质押比例达到50%或以上" else: continue rows.append({ "security_id": row.security_id, "source_code": row.source_code, "rule_code": code, "severity": severity, "description": description, "metric_value": row.pledge_ratio_pct, "metric_name": "pledge_ratio_pct", }) if rows: additions.append(pd.DataFrame(rows)) if not additions: return flags return pd.concat([flags, *additions], ignore_index=True, sort=False) def _score(frame: pd.DataFrame, weights: dict[str, float], flags: pd.DataFrame) -> pd.DataFrame: result = frame.copy() specs = { "profitability": [("weighted_roe_reported", True), ("net_margin", True), ("gross_margin_reported", True), ("roa_proxy", True)], "growth": [("operating_revenue_yoy_reported", True), ("net_profit_yoy_reported", True), ("revenue_cagr_3y", True), ("parent_profit_cagr_3y", True)], "cashflow": [("cfo_to_parent_profit", True), ("cfo_margin", True), ("cfo_ttm_yoy", True)], "balance_sheet": [("debt_ratio", False), ("current_ratio", True), ("quick_ratio", True), ("cash_short_debt_ratio", True)], "efficiency": [("asset_turnover_reported", True), ("receivables_turnover_reported", True), ("inventory_turnover_reported", True)], "industry_relative": [("weighted_roe_reported", True), ("operating_revenue_yoy_reported", True), ("gross_margin_reported", True)], "valuation": [("pe_ttm", False), ("pb", False), ("ps_ttm", False)], } score_columns = [] coverage_columns = [] for name, metrics in specs.items(): score, coverage = _component(result, metrics, float(weights[name])) score_col = f"{name}_score" coverage_col = f"{name}_coverage" result[score_col] = score result[coverage_col] = coverage score_columns.append(score_col) coverage_columns.append(coverage_col) result["score_coverage"] = result[coverage_columns].mean(axis=1) result["raw_total_score"] = result[score_columns].sum(axis=1, min_count=len(score_columns)) hard = flags.loc[flags["severity"].eq("hard")].groupby("source_code").size() warning = flags.loc[flags["severity"].eq("warning")].groupby("source_code").size() result["hard_risk_count"] = result["source_code"].map(hard).fillna(0).astype(int) result["warning_count"] = result["source_code"].map(warning).fillna(0).astype(int) result["risk_penalty"] = np.minimum(result["warning_count"] * 3, 15) result["total_score"] = result["raw_total_score"] - result["risk_penalty"] result.loc[result["hard_risk_count"].gt(0), "total_score"] = np.nan result.loc[result["is_financial_sector"], ["raw_total_score", "total_score"]] = np.nan result["financial_health_status"] = np.select( [result["hard_risk_count"].gt(0), result["warning_count"].ge(2), result["warning_count"].eq(1)], ["EXCLUDED", "HIGH_RISK", "ATTENTION"], default="NORMAL", ) result.loc[result["is_financial_sector"], "financial_health_status"] = "SPECIAL_MODEL_PENDING" result["fundamental_grade"] = pd.cut( result["total_score"], [-np.inf, 50, 65, 75, 85, np.inf], labels=["弱", "一般", "较好", "优秀", "突出"], right=False, ).astype("string") result["fundamental_eligible"] = ( ~result["is_financial_sector"] & result["hard_risk_count"].eq(0) & result["score_coverage"].ge(0.6) & result["total_score"].ge(65) ) result["model_version"] = FUNDAMENTAL_MODEL_VERSION return result def _score_financial_sectors(result: pd.DataFrame, weights: dict[str, float], flags: pd.DataFrame) -> pd.DataFrame: financial = result["is_financial_sector"] specs = { "financial_profitability": (["weighted_roe_reported", "roa_proxy", "net_margin", "financial_core_margin"], True, float(weights["profitability"])), "financial_growth": (["operating_revenue_yoy_reported", "net_profit_yoy_reported", "revenue_cagr_3y", "parent_profit_cagr_3y"], True, float(weights["growth"])), "financial_capital_strength": (["equity_assets_ratio", "parent_equity_yoy", "financial_solvency", "financial_buffer"], True, float(weights["capital_strength"])), } coverage_parts = [] score_parts = [] for name, (columns, direction, weight) in specs.items(): score, coverage = _component(result, [(column, direction) for column in columns], weight) result[f"{name}_score"] = score result[f"{name}_coverage"] = coverage score_parts.append(f"{name}_score") coverage_parts.append(f"{name}_coverage") quality_score, quality_coverage = _component( result, [("deducted_profit_ratio", True), ("impairment_revenue_ratio", False)], float(weights["earnings_quality"]), ) valuation_score, valuation_coverage = _component( result, [("pe_ttm", False), ("pb", False)], float(weights["valuation"]), ) result["financial_earnings_quality_score"] = quality_score result["financial_earnings_quality_coverage"] = quality_coverage result["financial_valuation_score"] = valuation_score result["financial_valuation_coverage"] = valuation_coverage score_parts.extend(["financial_earnings_quality_score", "financial_valuation_score"]) coverage_parts.extend(["financial_earnings_quality_coverage", "financial_valuation_coverage"]) hard = flags.loc[flags["severity"].eq("hard")].groupby("source_code").size() warning = flags.loc[flags["severity"].eq("warning")].groupby("source_code").size() result.loc[financial, "hard_risk_count"] = result.loc[financial, "source_code"].map(hard).fillna(0).astype(int) result.loc[financial, "warning_count"] = result.loc[financial, "source_code"].map(warning).fillna(0).astype(int) result.loc[financial, "risk_penalty"] = np.minimum(result.loc[financial, "warning_count"] * 3, 15) financial_coverage = result[coverage_parts].mean(axis=1) financial_raw = result[score_parts].sum(axis=1, min_count=len(score_parts)) result.loc[financial, "score_coverage"] = financial_coverage.loc[financial] result.loc[financial, "raw_total_score"] = financial_raw.loc[financial] result.loc[financial, "total_score"] = financial_raw.loc[financial] - result.loc[financial, "risk_penalty"] result.loc[financial & result["hard_risk_count"].gt(0), "total_score"] = np.nan result.loc[financial, "financial_health_status"] = np.select( [result.loc[financial, "hard_risk_count"].gt(0), result.loc[financial, "warning_count"].ge(2), result.loc[financial, "warning_count"].eq(1)], ["EXCLUDED", "HIGH_RISK", "ATTENTION"], default="NORMAL", ) result.loc[financial, "fundamental_grade"] = pd.cut( result.loc[financial, "total_score"], [-np.inf, 50, 65, 75, 85, np.inf], labels=["弱", "一般", "较好", "优秀", "突出"], right=False, ).astype("string") result.loc[financial, "fundamental_eligible"] = ( result.loc[financial, "hard_risk_count"].eq(0) & result.loc[financial, "score_coverage"].ge(0.6) & result.loc[financial, "total_score"].ge(65) ) return result def build_fundamentals(settings: Settings) -> FundamentalBuildResult: rules = _load_rules(settings) master = pd.read_parquet(settings.processed_dir / "security_master.parquet") finance = _add_calculated_ttm(pd.read_parquet(settings.financial_pit_path)) asof_date = pd.Timestamp(master["last_date"].max()) current, previous_1y, previous_3y = _select_asof_finance(finance, asof_date) columns = [ "security_id", "source_code", "name", "latest_close", "industry_l1_code", "industry_l1_name", "industry_l2_code", "industry_l2_name", "basic_pool_eligible", "is_st_current", ] frame = master[columns].merge(current, on="source_code", how="inner") sector_path = settings.processed_dir / "financial_sector_metrics.parquet" if sector_path.exists(): sector = pd.read_parquet(sector_path).drop(columns=[ "secucode", "industry_l2_code", "industry_l2_name", "report_period", "notice_date" ], errors="ignore") frame = frame.merge(sector, on="source_code", how="left") for column in ("financial_core_margin", "financial_solvency", "financial_buffer"): if column not in frame: frame[column] = np.nan # Re-align history rows to the merged current-finance order. history_key = current[["source_code"]].copy() p1 = history_key.merge(previous_1y.drop(columns="report_period"), on="source_code", how="left") p3 = history_key.merge(previous_3y.drop(columns="report_period"), on="source_code", how="left") p1 = frame[["source_code"]].merge(p1, on="source_code", how="left") p3 = frame[["source_code"]].merge(p3, on="source_code", how="left") frame["is_financial_sector"] = frame["industry_l1_code"].isin(["X50", "X51"]) metrics = _add_metrics(frame, p1, p3) flags = _risk_flags(metrics, rules["risk"]) flags = _add_financial_risk_flags(metrics, flags, rules["financial_risk"]) flags = _add_external_risk_flags(metrics, flags, settings, rules["governance_risk"]) scored = _score(metrics, rules["weights"], flags) scored = _score_financial_sectors(scored, rules["financial_weights"], flags) metrics_path = settings.processed_dir / "fundamental_metrics.parquet" scores_path = settings.processed_dir / "fundamental_scores.parquet" flags_path = settings.processed_dir / "fundamental_risk_flags.csv" metric_columns = [ "security_id", "source_code", "report_period", "announcement_date", "available_trade_date", "debt_ratio", "current_ratio", "quick_ratio", "cash_short_debt_ratio", "goodwill_equity_ratio", "cfo_to_parent_profit", "cfo_margin", "net_margin", "roa_proxy", "research_development_ratio", "receivables_yoy", "inventory_yoy", "cfo_ttm_yoy", "revenue_cagr_3y", "parent_profit_cagr_3y", "parent_equity_yoy", "equity_assets_ratio", "deducted_profit_ratio", "impairment_revenue_ratio", "operating_revenue_ttm_calc", "parent_net_profit_ttm_calc", "deducted_net_profit_ttm_calc", "operating_cash_flow_ttm_calc", "capital_expenditure_ttm_calc", "market_cap", "pe_ttm", "pb", "ps_ttm", ] scored[metric_columns].to_parquet(metrics_path, index=False) scored.to_parquet(scores_path, index=False) scored.to_csv(settings.processed_dir / "fundamental_scores.csv", index=False, encoding="utf-8-sig") flags.to_csv(flags_path, index=False, encoding="utf-8-sig") database = settings.database_dir / "fundamental_mvp.duckdb" with duckdb.connect(str(database)) as connection: connection.register("scores_df", scored) connection.register("flags_df", flags) connection.execute("CREATE OR REPLACE TABLE fundamental_scores AS SELECT * FROM scores_df") connection.execute("CREATE OR REPLACE TABLE fundamental_risk_flags AS SELECT * FROM flags_df") _write_report(settings, scored, flags, asof_date) return FundamentalBuildResult(len(scored), int(scored["total_score"].notna().sum()), int(scored["hard_risk_count"].gt(0).sum()), scores_path) def _write_report(settings: Settings, scored: pd.DataFrame, flags: pd.DataFrame, asof_date: pd.Timestamp) -> None: grades = scored["fundamental_grade"].fillna("未评分").value_counts() grade_lines = "\n".join(f"| {name} | {int(count)} |" for name, count in grades.items()) risks = Counter(flags["rule_code"]) risk_lines = "\n".join(f"| {name} | {count} |" for name, count in risks.most_common()) top = scored.loc[scored["fundamental_eligible"]].nlargest(20, "total_score") top_lines = "\n".join( f"| {row.security_id} | {row.name} | {row.industry_l2_name} | {row.total_score:.2f} | {row.warning_count} |" for row in top.itertuples(index=False) ) report = f"""# M1 基本面评分与风险规则报告 模型版本：{FUNDAMENTAL_MODEL_VERSION} 数据截面：{asof_date.date().isoformat()} 参与财务合并：{len(scored)} 只 普通工商企业已评分：{int(scored['total_score'].notna().sum())} 只 金融企业专用模型已评分：{int(scored['is_financial_sector'].sum())} 只 触发财务硬伤：{int(scored['hard_risk_count'].gt(0).sum())} 只 基本面达到较好及以上且无硬伤：{int(scored['fundamental_eligible'].sum())} 只 ## 1. 评分分布 | 等级 | 股票数 | |---|---:| {grade_lines} 评分采用行业内分位数，当前是研究版初始权重，不代表已经通过收益回测。银行、证券、保险和多元金融按盈利、成长、资本强度、盈利质量和估值五个专用维度评分，不套用普通工商企业的现金流和流动比率规则。 营收、归母净利润和经营现金流 TTM 由累计报表还原单季后滚动四季计算，不使用单位口径混合的便利字段。 ## 2. 风险规则命中 | 规则 | 命中数 | |---|---:| {risk_lines} 审计意见、监管处罚、资金占用和质押数据尚未接入，因此“无现有规则命中”不等于财务绝对安全。 ## 3. 当前高分样本 | 代码 | 名称 | 二级行业 | 风险后得分 | 警告数 | |---|---|---|---:|---:| {top_lines} 高分样本仅用于检查评分逻辑和进入后续候选池，不构成投资建议。 """ (settings.report_dir / "M1_基本面评分报告.md").write_text(report, encoding="utf-8")