from __future__ import annotations import json from pathlib import Path from typing import Any import uuid import pandas as pd from .config import Settings FACTOR_LABELS = { "profitability": "盈利质量", "growth": "成长", "balance_sheet": "资产负债表", "cash_quality": "现金流", "book_to_price": "价值", "efficiency": "营运效率", "gross_margin": "毛利率", "liquidity": "流动性", "rd_intensity": "研发强度", } def _number(value: Any, digits: int = 2) -> float | None: parsed = pd.to_numeric(value, errors="coerce") return None if pd.isna(parsed) else round(float(parsed), digits) def _text(value: Any, fallback: str = "—") -> str: return fallback if pd.isna(value) or str(value).strip() == "" else str(value) def _read_csv(path: Path) -> pd.DataFrame: if not path.exists(): return pd.DataFrame() return pd.read_csv(path, low_memory=False) def build_dashboard_payload(settings: Settings) -> dict[str, Any]: processed = settings.processed_dir reports = settings.report_dir m1 = json.loads((reports / "m1_run_summary.json").read_text(encoding="utf-8")) m2 = json.loads((reports / "m2_run_summary.json").read_text(encoding="utf-8")) industries = _read_csv(processed / "industry_hotspot_scores.csv") candidates = _read_csv(processed / "m2_stock_candidates.csv") cards = _read_csv(processed / "stock_health_cards.csv") risks = _read_csv(processed / "external_governance_risk_flags.csv") factor_summary = _read_csv(processed / "factor_backtest_summary.csv") factor_ic = _read_csv(processed / "factor_ic.csv") stock_momentum = _read_csv(processed / "stock_momentum_snapshot.csv") full_market = _read_csv(processed / "full_market_catalog.csv") if not candidates.empty: eligible = candidates[candidates["candidate_eligible"].fillna(False).astype(bool)].copy() eligible = eligible.sort_values( ["candidate_tier_order", "m2_composite_score", "combined_score"], ascending=[True, False, False], na_position="last", ) else: eligible = candidates card_columns = [ "source_code", "quant_highlights", "risk_codes", "risk_notes", "pledge_ratio_pct", "weighted_roe_reported", "gross_margin_reported", "operating_revenue_yoy_reported", "net_profit_yoy_reported", "cfo_to_parent_profit", "pe_ttm", "pb", ] available_card_columns = [column for column in card_columns if column in cards.columns] if not eligible.empty and available_card_columns: eligible = eligible.merge(cards[available_card_columns], on="source_code", how="left", suffixes=("", "_card")) candidate_rows = [] score_fields = [ ("profitability_score", "盈利", 20), ("growth_score", "成长", 20), ("cashflow_score", "现金流", 20), ("balance_sheet_score", "资产负债", 15), ("efficiency_score", "效率", 10), ("valuation_score", "估值", 5), ] for _, row in eligible.iterrows(): hotspot_value = row.get("hotspot_score") if pd.isna(hotspot_value): hotspot_value = row.get("hotspot_score_x") if pd.isna(hotspot_value): hotspot_value = row.get("hotspot_score_y") pb_value = _number(row.get("pb"), 4) dimensions = [ {"label": label, "value": _number(row.get(field)), "max": maximum} for field, label, maximum in score_fields ] candidate_rows.append({ "code": _text(row.get("source_code")), "securityId": _text(row.get("security_id")), "name": _text(row.get("name")), "industry": _text(row.get("industry_l2_name")), "industryL1": _text(row.get("industry_l1_name")), "tier": _text(row.get("candidate_tier")), "score": _number(row.get("m2_composite_score") if pd.notna(row.get("m2_composite_score")) else row.get("combined_score")), "fundamental": _number(row.get("total_score")), "hotspot": _number(hotspot_value), "close": _number(row.get("latest_close")), "warnings": int(_number(row.get("warning_count"), 0) or 0), "hardRisks": int(_number(row.get("hard_risk_count"), 0) or 0), "health": _text(row.get("financial_health_status")), "reportPeriod": _text(row.get("report_period")), "highlights": _text(row.get("quant_highlights"), "暂无摘要"), "riskNotes": _text(row.get("risk_notes"), "未命中已建量化风险规则"), "pledgeRatio": _number(row.get("pledge_ratio_pct")), "roe": _number(row.get("weighted_roe_reported")), "grossMargin": _number(row.get("gross_margin_reported")), "revenueYoy": _number(row.get("operating_revenue_yoy_reported")), "profitYoy": _number(row.get("net_profit_yoy_reported")), "cfoProfit": _number(row.get("cfo_to_parent_profit")), "pe": _number(row.get("pe_ttm")), "pb": _number(row.get("pb")), "researchCoverage": _number(row.get("m2_research_coverage")), "factorValues": { "profitability": _number(row.get("profitability_score"), 4), "growth": _number(row.get("growth_score"), 4), "balance_sheet": _number(row.get("balance_sheet_score"), 4), "cash_quality": _number(row.get("cashflow_score"), 4), "book_to_price": round(1 / pb_value, 4) if pb_value and pb_value > 0 else None, "efficiency": _number(row.get("efficiency_score"), 4), "gross_margin": _number(row.get("gross_margin_reported"), 4), "liquidity": _number(row.get("current_ratio"), 4), "rd_intensity": _number(row.get("research_development_ratio"), 4), }, "dimensions": dimensions, }) hot_industries = [] if not industries.empty: ordered = industries[pd.to_numeric(industries["hotspot_score"], errors="coerce").notna()].sort_values("hotspot_score", ascending=False) for _, row in ordered.head(12).iterrows(): hot_industries.append({ "name": _text(row.get("industry_l2_name")), "parent": _text(row.get("industry_l1_name")), "score": _number(row.get("hotspot_score")), "return20": _number(row.get("return_20d_median")), "return10": _number(row.get("return_10d_median")), "breadth": _number(float(row.get("positive_20d_ratio", 0)) * 100), "activity": _number(row.get("amount_ratio_5_20_median")), "members": int(_number(row.get("member_count"), 0) or 0), }) momentum_10d = [] if not industries.empty and "return_10d_median" in industries.columns: valid_momentum = industries[ pd.to_numeric(industries["return_10d_median"], errors="coerce").notna() & industries["member_coverage_ok"].fillna(False).astype(bool) ].sort_values("return_10d_median", ascending=False) for rank, (_, row) in enumerate(valid_momentum.head(10).iterrows(), start=1): momentum_10d.append({ "rank": rank, "name": _text(row.get("industry_l2_name")), "parent": _text(row.get("industry_l1_name")), "return10": _number(row.get("return_10d_median")), "return5": _number(row.get("return_5d_median")), "return20": _number(row.get("return_20d_median")), "members": int(_number(row.get("member_count"), 0) or 0), }) momentum_20d = [] if not industries.empty and "return_20d_median" in industries.columns: valid_momentum = industries[ pd.to_numeric(industries["return_20d_median"], errors="coerce").notna() & industries["member_coverage_ok"].fillna(False).astype(bool) ].sort_values("return_20d_median", ascending=False) for rank, (_, row) in enumerate(valid_momentum.head(10).iterrows(), start=1): momentum_20d.append({ "rank": rank, "name": _text(row.get("industry_l2_name")), "parent": _text(row.get("industry_l1_name")), "return20": _number(row.get("return_20d_median")), "return10": _number(row.get("return_10d_median")), "return5": _number(row.get("return_5d_median")), "members": int(_number(row.get("member_count"), 0) or 0), }) industry_details = [] if not industries.empty: momentum_with_scores = stock_momentum if not stock_momentum.empty and not candidates.empty: score_columns = [ column for column in ["source_code", "m2_composite_score", "total_score", "candidate_tier", "candidate_eligible"] if column in candidates.columns ] momentum_with_scores = stock_momentum.merge( candidates[score_columns].drop_duplicates("source_code"), on="source_code", how="left" ) def leaders_for(industry_name: str, period: int) -> list[dict[str, Any]]: if momentum_with_scores.empty: return [] return_column = f"return_{period}d" rows = momentum_with_scores[ momentum_with_scores["industry_l2_name"].eq(industry_name) & pd.to_numeric(momentum_with_scores.get(return_column), errors="coerce").notna() & ~momentum_with_scores["is_st_current"].fillna(False).astype(bool) ].sort_values(return_column, ascending=False).head(6) return [{ "code": _text(row.get("source_code")).upper(), "name": _text(row.get("name")), "return": _number(row.get(return_column)), "amountRatio": _number(row.get("amount_ratio_5_20")), "score": _number(row.get("m2_composite_score")), "tier": _text(row.get("candidate_tier"), "未入候选"), } for _, row in rows.iterrows()] for _, row in industries.iterrows(): name = _text(row.get("industry_l2_name")) industry_details.append({ "name": name, "parent": _text(row.get("industry_l1_name")), "members": int(_number(row.get("member_count"), 0) or 0), "hotspotScore": _number(row.get("hotspot_score")), "hotspotRank": int(_number(row.get("hotspot_rank"), 0) or 0), "rank10": int(_number(row.get("momentum_10d_rank"), 0) or 0), "rank20": int(_number(row.get("momentum_20d_rank"), 0) or 0), "return5": _number(row.get("return_5d_median")), "return10": _number(row.get("return_10d_median")), "return20": _number(row.get("return_20d_median")), "return60": _number(row.get("return_60d_median")), "positive10": _number(float(row.get("positive_10d_ratio", 0)) * 100), "positive20": _number(float(row.get("positive_20d_ratio", 0)) * 100), "aboveMa20": _number(float(row.get("above_ma20_ratio", 0)) * 100), "nearHigh60": _number(float(row.get("near_high60_ratio", 0)) * 100), "amountRatio": _number(row.get("amount_ratio_5_20_median")), "trendAlignment": _number(float(row.get("trend_alignment", 0)) * 100), "components": [ {"label": "价格强度", "value": _number(row.get("momentum_score")), "max": 35}, {"label": "上涨扩散", "value": _number(row.get("breadth_score")), "max": 25}, {"label": "成交活跃", "value": _number(row.get("activity_score")), "max": 20}, {"label": "趋势持续", "value": _number(row.get("persistence_score")), "max": 20}, ], "leaders10": leaders_for(name, 10), "leaders20": leaders_for(name, 20), "evidenceStatus": "QUANTITATIVE_ONLY", }) factor_rows = [] if not factor_summary.empty: for _, row in factor_summary.sort_values("mean_rank_ic", ascending=False).iterrows(): factor = _text(row.get("factor")) factor_rows.append({ "key": factor, "label": FACTOR_LABELS.get(factor, factor), "rankIc": _number(row.get("mean_rank_ic"), 4), "positiveRate": _number(float(row.get("ic_positive_rate", 0)) * 100), "cumulative": _number(float(row.get("cumulative_long_short_net_return", 0)) * 100), }) ic_series: dict[str, list[dict[str, Any]]] = {} if not factor_ic.empty: factor_ic["rebalance_date"] = pd.to_datetime(factor_ic["rebalance_date"], errors="coerce") cutoff = factor_ic["rebalance_date"].max() - pd.DateOffset(months=35) recent_ic = factor_ic[factor_ic["rebalance_date"] >= cutoff] for factor in ["book_to_price", "growth", "cash_quality", "profitability"]: rows = recent_ic[recent_ic["factor"] == factor].sort_values("rebalance_date") ic_series[factor] = [ {"date": row.rebalance_date.strftime("%Y-%m"), "value": _number(row.rank_ic, 4)} for row in rows.itertuples(index=False) ] risk_events = [] if not risks.empty: risks["event_date"] = pd.to_datetime(risks["event_date"], errors="coerce") latest = risks.sort_values(["event_date", "severity"], ascending=[False, True]).head(16) for _, row in latest.iterrows(): risk_events.append({ "date": row["event_date"].strftime("%m-%d") if pd.notna(row["event_date"]) else "—", "code": _text(row.get("source_code")).upper(), "severity": _text(row.get("severity")), "type": _text(row.get("rule_code")), "description": _text(row.get("description")), "url": _text(row.get("source_url"), ""), }) tier_counts = {} if not eligible.empty: tier_counts = {str(key): int(value) for key, value in eligible["candidate_tier"].value_counts().items()} hard_count = int((risks.get("severity", pd.Series(dtype=str)) == "hard").sum()) warning_count = int((risks.get("severity", pd.Series(dtype=str)) == "warning").sum()) # ---- 全市场目录（覆盖全部个股，只展示不筛选）------------------------ # 数据来自 deploy/build_full_market_catalog.py 的产出。缺文件时静默降级， # 不影响看板其余部分生成。 full_market_counts: dict[str, int] = {} full_market_rows: list[dict[str, Any]] = [] if not full_market.empty and "catalog_tier" in full_market.columns: fm = full_market total_n = len(fm) full_market_counts = { "total": int(total_n), "basicPool": int(fm.get("in_basic_pool", pd.Series(dtype=bool)).fillna(False).astype(bool).sum()), "candidatePool": int(fm.get("in_candidate_pool", pd.Series(dtype=bool)).fillna(False).astype(bool).sum()), "tiers": {str(k): int(v) for k, v in fm["catalog_tier"].value_counts().items()}, } reason_text = fm.get("exclude_reasons", pd.Series("", index=fm.index)).fillna("").astype(str) def _flag(column: str) -> list[bool]: if column not in fm.columns: return [False] * total_n return fm[column].fillna(False).astype(bool).tolist() codes = fm.get("source_code", pd.Series("", index=fm.index)).fillna("").astype(str).tolist() names = fm.get("name", pd.Series("", index=fm.index)).fillna("").astype(str).tolist() tiers = fm["catalog_tier"].fillna("").astype(str).tolist() ind1 = fm.get("industry_l1_name", pd.Series("", index=fm.index)).fillna("").astype(str).tolist() ind2 = fm.get("industry_l2_name", pd.Series("", index=fm.index)).fillna("").astype(str).tolist() in_basic = _flag("in_basic_pool") in_fund = _flag("in_fundamental_eligible") in_cand = _flag("in_candidate_pool") is_st = _flag("is_st_current") is_fin = _flag("is_financial_sector") numbers = {} for key, digits in ( ("total_score", 2), ("hotspot_score", 2), ("market_cap", 0), ("pe_ttm", 2), ("pb", 2), ("latest_close", 2), ("avg_amount_20", 0), ): series = fm[key] if key in fm.columns else pd.Series([None] * total_n, index=fm.index) numbers[key] = [_number(value, digits) for value in series] for i in range(total_n): full_market_rows.append({ "code": codes[i].upper(), "name": names[i], "tier": tiers[i], "ind1": ind1[i], "ind2": ind2[i], "score": numbers["total_score"][i], "hot": numbers["hotspot_score"][i], "cap": numbers["market_cap"][i], "pe": numbers["pe_ttm"][i], "pb": numbers["pb"][i], "close": numbers["latest_close"][i], "amt": numbers["avg_amount_20"][i], "basic": in_basic[i], "fund": in_fund[i], "cand": in_cand[i], "st": is_st[i], "fin": is_fin[i], "reason": reason_text.iloc[i], }) return { "runId": uuid.uuid4().hex, "generatedAt": pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S"), "asOf": _text(m1.get("market_data_cutoff")), "factorAsOf": _text(m2.get("as_of")), "summary": { "universe": int(m1.get("finance_covered_stocks", 0)), "eligible": int(len(eligible)), "hotIndustries": int(m1.get("hotspot_build", {}).get("hot_industries", 0)), "hardRisks": hard_count, "warningRisks": warning_count, "financeCoverage": round(float(m1.get("finance_covered_stocks", 0)) / max(float(m1.get("named_stocks", 1)), 1) * 100, 1), "staleDays": int(m1.get("market_data_business_days_stale", 0)), "tierCounts": tier_counts, }, "latestWeights": [ {"key": key, "label": FACTOR_LABELS.get(key, key), "value": round(float(value) * 100, 1)} for key, value in m2.get("latest_weights", {}).items() ], "industries": hot_industries, "momentum10": momentum_10d, "momentum20": momentum_20d, "industryDetails": industry_details, "candidates": candidate_rows, "factors": factor_rows, "icSeries": ic_series, "risks": risk_events, "limitations": list(m1.get("limitations", [])), "fullMarket": full_market_rows, "fullMarketCounts": full_market_counts, } def write_dashboard(settings: Settings, output_path: Path | None = None) -> Path: template_path = Path(__file__).with_name("assets") / "dashboard.html" output_path = output_path or settings.report_dir / "dashboard" / "index.html" output_path.parent.mkdir(parents=True, exist_ok=True) payload = build_dashboard_payload(settings) payload_json = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).replace("