from __future__ import annotations from dataclasses import dataclass from pathlib import Path import tomllib import duckdb import numpy as np import pandas as pd from .config import Settings HOTSPOT_MODEL_VERSION = "0.1.0" @dataclass(frozen=True) class HotspotBuildResult: industries: int hot_industries: int candidates: int output_path: Path def _pct_rank(series: pd.Series) -> pd.Series: return pd.to_numeric(series, errors="coerce").rank(pct=True) def _load_recent_stock_metrics(settings: Settings, asof_date: pd.Timestamp) -> pd.DataFrame: start = (asof_date - pd.Timedelta(days=150)).date().isoformat() path = str(settings.daily_price_path).replace("'", "''") query = f""" WITH recent AS ( SELECT source_code, trade_date, close_qfq, amount_yuan, row_number() OVER (PARTITION BY source_code ORDER BY trade_date DESC) AS rn FROM read_parquet('{path}') WHERE trade_date >= DATE '{start}' AND trade_date <= DATE '{asof_date.date().isoformat()}' AND close_qfq > 0 ), points AS ( SELECT source_code, max(CASE WHEN rn=1 THEN trade_date END) latest_trade_date, max(CASE WHEN rn=1 THEN close_qfq END) close_latest, max(CASE WHEN rn=6 THEN close_qfq END) close_5, max(CASE WHEN rn=11 THEN close_qfq END) close_10, max(CASE WHEN rn=21 THEN close_qfq END) close_20, max(CASE WHEN rn=61 THEN close_qfq END) close_60, avg(amount_yuan) FILTER (WHERE rn<=5) amount_avg_5, avg(amount_yuan) FILTER (WHERE rn<=20) amount_avg_20, avg(close_qfq) FILTER (WHERE rn<=20) close_ma20, max(close_qfq) FILTER (WHERE rn<=60) close_high_60, count(*) FILTER (WHERE rn<=61) observations FROM recent GROUP BY source_code ) SELECT *, (close_latest/close_5-1)*100 AS return_5d, (close_latest/close_10-1)*100 AS return_10d, (close_latest/close_20-1)*100 AS return_20d, (close_latest/close_60-1)*100 AS return_60d, amount_avg_5/nullif(amount_avg_20,0) AS amount_ratio_5_20, close_latest/close_ma20-1 AS above_ma20_pct, close_latest/close_high_60 AS high_60_ratio FROM points """ with duckdb.connect() as connection: return connection.execute(query).fetchdf() def _score_industries(stock: pd.DataFrame, settings: Settings) -> pd.DataFrame: rules = tomllib.loads((settings.project_root / "config" / "fundamental_rules.toml").read_text(encoding="utf-8")) config = rules["hotspot"] eligible = stock.loc[ stock["industry_l2_code"].notna() & ~stock["is_st_current"] & stock["observations"].ge(61) ].copy() eligible["is_positive_20d"] = eligible["return_20d"].gt(0) eligible["is_positive_10d"] = eligible["return_10d"].gt(0) eligible["is_above_ma20"] = eligible["above_ma20_pct"].gt(0) eligible["is_near_high60"] = eligible["high_60_ratio"].ge(0.95) eligible["trend_alignment"] = eligible[["return_5d", "return_20d", "return_60d"]].gt(0).sum(axis=1) / 3 group_columns = ["industry_l1_code", "industry_l1_name", "industry_l2_code", "industry_l2_name"] industry = eligible.groupby(group_columns, dropna=False).agg( member_count=("source_code", "size"), return_5d_median=("return_5d", "median"), return_10d_median=("return_10d", "median"), return_20d_median=("return_20d", "median"), return_60d_median=("return_60d", "median"), positive_10d_ratio=("is_positive_10d", "mean"), positive_20d_ratio=("is_positive_20d", "mean"), above_ma20_ratio=("is_above_ma20", "mean"), near_high60_ratio=("is_near_high60", "mean"), amount_ratio_5_20_median=("amount_ratio_5_20", "median"), amount_avg_20_total=("amount_avg_20", "sum"), trend_alignment=("trend_alignment", "mean"), ).reset_index() industry["member_coverage_ok"] = industry["member_count"].ge(int(config["minimum_members"])) industry["amount_scale"] = np.log1p(industry["amount_avg_20_total"].clip(lower=0)) industry["momentum_score"] = pd.concat([ _pct_rank(industry["return_5d_median"]), _pct_rank(industry["return_20d_median"]), _pct_rank(industry["return_60d_median"]), ], axis=1).mean(axis=1) * float(config["momentum_weight"]) industry["breadth_score"] = pd.concat([ _pct_rank(industry["positive_20d_ratio"]), _pct_rank(industry["above_ma20_ratio"]), _pct_rank(industry["near_high60_ratio"]), ], axis=1).mean(axis=1) * float(config["breadth_weight"]) industry["activity_score"] = pd.concat([ _pct_rank(industry["amount_ratio_5_20_median"]), _pct_rank(industry["amount_scale"]), ], axis=1).mean(axis=1) * float(config["activity_weight"]) industry["persistence_score"] = pd.concat([ _pct_rank(industry["trend_alignment"]), _pct_rank(industry["return_20d_median"]), _pct_rank(industry["return_60d_median"]), ], axis=1).mean(axis=1) * float(config["persistence_weight"]) industry["hotspot_score"] = industry[["momentum_score", "breadth_score", "activity_score", "persistence_score"]].sum(axis=1) industry.loc[~industry["member_coverage_ok"], "hotspot_score"] = np.nan industry["hotspot_rank"] = industry["hotspot_score"].rank(ascending=False, method="min").astype("Int64") industry["momentum_10d_rank"] = industry["return_10d_median"].rank(ascending=False, method="min").astype("Int64") industry["momentum_20d_rank"] = industry["return_20d_median"].rank(ascending=False, method="min").astype("Int64") industry["is_hot"] = industry["hotspot_score"].ge(float(config["hot_score_threshold"])) industry["model_version"] = HOTSPOT_MODEL_VERSION return industry.sort_values("hotspot_score", ascending=False, na_position="last") def _build_candidates(fundamentals: pd.DataFrame, industry: pd.DataFrame, asof_date: pd.Timestamp, settings: Settings) -> pd.DataFrame: rules = tomllib.loads((settings.project_root / "config" / "fundamental_rules.toml").read_text(encoding="utf-8")) config = rules["candidate"] hot = industry[["industry_l2_code", "hotspot_score", "hotspot_rank", "is_hot"]] frame = fundamentals.merge(hot, on="industry_l2_code", how="left") quality_path = settings.processed_dir / "quality_signal_proxies.parquet" if quality_path.exists(): quality = pd.read_parquet(quality_path).drop(columns=[ "security_id", "name", "industry_l2_code", "industry_l2_name", "model_version" ], errors="ignore") frame = frame.merge(quality, on="source_code", how="left") frame["combined_score"] = frame["total_score"] * float(config["fundamental_weight"]) + frame["hotspot_score"] * float(config["hotspot_weight"]) frame["candidate_tier"] = np.select([ frame["fundamental_eligible"] & frame["total_score"].ge(75) & frame["hotspot_score"].ge(65), frame["fundamental_eligible"] & frame["hotspot_score"].ge(75), frame["fundamental_eligible"] & frame["total_score"].ge(75), frame["fundamental_eligible"], ], ["优先研究", "热点交集", "基本面精选", "观察"], default="排除") frame["candidate_eligible"] = frame["fundamental_eligible"] & frame["basic_pool_eligible"] today = pd.Timestamp.today().normalize() staleness = int(np.busday_count(asof_date.date(), today.date())) if asof_date < today else 0 frame["market_snapshot_date"] = asof_date frame["market_data_business_days_stale"] = staleness frame["is_actionable_snapshot"] = staleness <= settings.max_business_day_staleness order = {"优先研究": 1, "热点交集": 2, "基本面精选": 3, "观察": 4, "排除": 5} frame["candidate_tier_order"] = frame["candidate_tier"].map(order) return frame.sort_values(["candidate_eligible", "candidate_tier_order", "combined_score"], ascending=[False, True, False]) def _build_health_cards(candidates: pd.DataFrame, settings: Settings) -> pd.DataFrame: flags = pd.read_csv(settings.processed_dir / "fundamental_risk_flags.csv") if flags.empty: cautions = pd.DataFrame(columns=["source_code", "risk_codes", "risk_notes"]) else: aggregations = { "risk_codes": ("rule_code", lambda x: ";".join(dict.fromkeys(x.astype(str)))), "risk_notes": ("description", lambda x: "；".join(dict.fromkeys(x.astype(str)))), } if "source_url" in flags.columns: aggregations["risk_source_urls"] = ( "source_url", lambda x: ";".join(dict.fromkeys(x.dropna().astype(str))) ) cautions = flags.groupby("source_code").agg(**aggregations).reset_index() cards = candidates.merge(cautions, on="source_code", how="left") pledge_path = settings.processed_dir / "external_pledge_latest.parquet" if pledge_path.exists(): pledge = pd.read_parquet(pledge_path)[["source_code", "pledge_date", "pledge_ratio_pct", "pledge_deal_count"]] cards = cards.merge(pledge, on="source_code", how="left") cards["pledge_record_status"] = np.where(cards["pledge_ratio_pct"].notna(), "CURRENT_RECORD_FOUND", "NO_CURRENT_RECORD") cards["governance_data_status"] = "AVAILABLE" if pledge_path.exists() else "NOT_REFRESHED" def highlights(row: pd.Series) -> str: if row.get("is_financial_sector", False): tests = [ (row.get("financial_profitability_score", np.nan) >= 21, "金融盈利能力行业前列"), (row.get("financial_growth_score", np.nan) >= 17.5, "金融成长表现行业前列"), (row.get("financial_capital_strength_score", np.nan) >= 14, "资本强度行业前列"), (row.get("financial_earnings_quality_score", np.nan) >= 10.5, "盈利质量行业前列"), (row.get("financial_valuation_score", np.nan) >= 7, "估值行业内相对有吸引力"), (row.get("hotspot_score", np.nan) >= 70, "所属行业当前较热"), ] else: tests = [ (row.get("profitability_score", np.nan) >= 14, "盈利能力行业前列"), (row.get("growth_score", np.nan) >= 14, "成长表现行业前列"), (row.get("cashflow_score", np.nan) >= 14, "现金流质量较好"), (row.get("balance_sheet_score", np.nan) >= 10.5, "资产负债表相对稳健"), (row.get("efficiency_score", np.nan) >= 7, "营运效率行业前列"), (row.get("valuation_score", np.nan) >= 3.5, "估值行业内相对有吸引力"), (row.get("hotspot_score", np.nan) >= 70, "所属行业当前较热"), ] notes = [text for passed, text in tests if passed] return "；".join(notes) if notes else "暂无量化突出项" cards["quant_highlights"] = cards.apply(highlights, axis=1) cards["risk_codes"] = cards["risk_codes"].fillna("") cards["risk_notes"] = cards["risk_notes"].fillna("未命中已建量化风险规则") if "risk_source_urls" in cards: cards["risk_source_urls"] = cards["risk_source_urls"].fillna("") cards["audit_verification_status"] = np.select([ cards["risk_codes"].str.contains("AUDIT_ADVERSE"), cards["risk_codes"].str.contains("AUDIT_QUALIFIED"), ], ["KNOWN_ADVERSE_OR_DISCLAIMER", "KNOWN_QUALIFIED_OR_NONSTANDARD"], default="NOT_VERIFIED_BY_CURRENT_DATASET") card_columns = [ "security_id", "source_code", "name", "industry_l1_name", "industry_l2_name", "report_period", "announcement_date", "available_trade_date", "market_snapshot_date", "financial_health_status", "hard_risk_count", "warning_count", "risk_codes", "risk_notes", "risk_source_urls", "governance_data_status", "audit_verification_status", "pledge_record_status", "pledge_date", "pledge_ratio_pct", "pledge_deal_count", "quant_highlights", "fundamental_grade", "total_score", "score_coverage", "profitability_score", "growth_score", "cashflow_score", "balance_sheet_score", "efficiency_score", "valuation_score", "debt_ratio", "current_ratio", "cash_short_debt_ratio", "financial_profitability_score", "financial_growth_score", "financial_capital_strength_score", "financial_earnings_quality_score", "financial_valuation_score", "financial_core_margin", "financial_solvency", "financial_buffer", "net_interest_margin", "provision_coverage_ratio", "capital_adequacy_ratio", "risk_coverage_ratio", "net_capital_liabilities_ratio", "solvency_adequacy_ratio", "moat_quant_proxy_score", "moat_proxy_level", "innovation_quant_proxy_score", "innovation_proxy_level", "management_quant_proxy_score", "quality_signal_coverage", "profitable_years_5y", "roe_above_10_years_5y", "gross_margin_stability_5y", "evidence_status", "moat_evidence_status", "goodwill_equity_ratio", "weighted_roe_reported", "gross_margin_reported", "operating_revenue_yoy_reported", "net_profit_yoy_reported", "revenue_cagr_3y", "parent_profit_cagr_3y", "cfo_to_parent_profit", "pe_ttm", "pb", "ps_ttm", "hotspot_score", "hotspot_rank", "candidate_tier", "combined_score", "candidate_eligible", "is_actionable_snapshot", "market_data_business_days_stale", "model_version", ] return cards[[column for column in card_columns if column in cards.columns]] def build_hotspots(settings: Settings) -> HotspotBuildResult: master = pd.read_parquet(settings.processed_dir / "security_master.parquet") fundamentals = pd.read_parquet(settings.processed_dir / "fundamental_scores.parquet") asof_date = pd.Timestamp(master["last_date"].max()) recent = _load_recent_stock_metrics(settings, asof_date) stock_momentum = master[[ "security_id", "source_code", "name", "industry_l1_code", "industry_l1_name", "industry_l2_code", "industry_l2_name", "is_st_current", "basic_pool_eligible", ]].merge(recent, on="source_code", how="left") industry = _score_industries(stock_momentum, settings) candidates = _build_candidates(fundamentals, industry, asof_date, settings) cards = _build_health_cards(candidates, settings) industry_path = settings.processed_dir / "industry_hotspot_scores.parquet" candidate_path = settings.processed_dir / "stock_screen_candidates.parquet" industry.to_parquet(industry_path, index=False) industry.to_csv(settings.processed_dir / "industry_hotspot_scores.csv", index=False, encoding="utf-8-sig") stock_momentum.to_parquet(settings.processed_dir / "stock_momentum_snapshot.parquet", index=False) stock_momentum.to_csv(settings.processed_dir / "stock_momentum_snapshot.csv", index=False, encoding="utf-8-sig") candidates.to_parquet(candidate_path, index=False) candidates.to_csv(settings.processed_dir / "stock_screen_candidates.csv", index=False, encoding="utf-8-sig") cards.to_parquet(settings.processed_dir / "stock_health_cards.parquet", index=False) cards.to_csv(settings.processed_dir / "stock_health_cards.csv", index=False, encoding="utf-8-sig") with duckdb.connect(str(settings.database_dir / "fundamental_mvp.duckdb")) as connection: connection.register("industry_df", industry) connection.register("candidates_df", candidates) connection.register("cards_df", cards) connection.register("stock_momentum_df", stock_momentum) connection.execute("CREATE OR REPLACE TABLE industry_hotspot_scores AS SELECT * FROM industry_df") connection.execute("CREATE OR REPLACE TABLE stock_screen_candidates AS SELECT * FROM candidates_df") connection.execute("CREATE OR REPLACE TABLE stock_health_cards AS SELECT * FROM cards_df") connection.execute("CREATE OR REPLACE TABLE stock_momentum_snapshot AS SELECT * FROM stock_momentum_df") _write_report(settings, industry, candidates, asof_date) return HotspotBuildResult(len(industry), int(industry["is_hot"].sum()), int(candidates["candidate_eligible"].sum()), candidate_path) def _write_report(settings: Settings, industry: pd.DataFrame, candidates: pd.DataFrame, asof_date: pd.Timestamp) -> None: industry_lines = "\n".join( f"| {int(row.hotspot_rank)} | {row.industry_l2_name} | {row.member_count} | {row.hotspot_score:.2f} | {row.return_20d_median:.2f}% | {row.positive_20d_ratio:.1%} | {row.amount_ratio_5_20_median:.2f} |" for row in industry.head(20).itertuples(index=False) if pd.notna(row.hotspot_score) ) selected = candidates.loc[candidates["candidate_eligible"]].head(50) candidate_lines = "\n".join( f"| {row.security_id} | {row.name} | {row.industry_l2_name} | {row.candidate_tier} | {row.total_score:.2f} | {row.hotspot_score:.2f} | {row.combined_score:.2f} |" for row in selected.itertuples(index=False) ) stale = int(candidates["market_data_business_days_stale"].iloc[0]) report = f"""# M1 行业热点与候选股池报告 热点模型版本：{HOTSPOT_MODEL_VERSION} 市场截面：{asof_date.date().isoformat()} 截面滞后：{stale} 个工作日 二级行业：{len(industry)} 个 热点得分达 70：{int(industry['is_hot'].sum())} 个 通过基础证券池与基本面门槛：{int(candidates['candidate_eligible'].sum())} 只 > 当前行情已超过时效门槛，本报告只能用于模型验证和历史截面研究，不能作为当日交易依据。 ## 1. 二级行业热度榜 | 排名 | 二级行业 | 成分数 | 热度 | 20日中位涨幅 | 上涨扩散度 | 5/20日量能比 | |---:|---|---:|---:|---:|---:|---:| {industry_lines} 热度由价格强度 35%、上涨扩散度 25%、成交活跃度 20%、趋势持续性 20% 合成，均为当前 128 个二级行业的横截面分位得分。 ## 2. 初版候选股池 | 代码 | 名称 | 二级行业 | 分层 | 基本面 | 行业热度 | 综合分 | |---|---|---|---|---:|---:|---:| {candidate_lines} 综合分为基本面 70% + 行业热度 30%。候选股必须同时通过上市时长、流动性、ST/退市标签、数据新鲜度、财务硬伤和基本面得分门槛。 """ (settings.report_dir / "M1_行业热点与候选股池.md").write_text(report, encoding="utf-8")