[codex] add chinext growth momentum quality snapshot

README.md

@@ -15,6 +15,7 @@ This repository produces evidence and artifacts. It does not place broker orders
 | Profile | Display name | Contract | Builder |
 | --- | --- | --- | --- |
 | `cn_dividend_quality_snapshot` | CN Dividend Quality Snapshot | `cn_dividend_quality_snapshot.factor_snapshot.v1` | `cneq-build-dividend-quality-snapshot` |
+| `cn_chinext_growth_momentum_quality_snapshot` | CN ChiNext Growth Momentum Quality Snapshot | `cn_chinext_growth_momentum_quality_snapshot.factor_snapshot.v1` | `cneq-build-chinext-growth-momentum-quality-snapshot` |
 
 ## Quick start
 
@@ -27,6 +28,7 @@ Build a sample artifact pack locally:
 
 ```bash
 PYTHONPATH=src python scripts/build_dividend_quality_sample.py
+PYTHONPATH=src python scripts/build_chinext_growth_momentum_quality_sample.py
 ```
 
 Stage real factor inputs via AkShare (falls back to sample CSV on failure):
@@ -42,6 +44,10 @@ Or use the installed entrypoint:
 cneq-build-dividend-quality-snapshot \
   --factor-snapshot examples/dividend_quality/factor_snapshot.sample.csv \
   --output-dir data/output/dividend_quality
+
+cneq-build-chinext-growth-momentum-quality-snapshot \
+  --factor-snapshot examples/chinext_growth_momentum_quality/factor_snapshot.sample.csv \
+  --output-dir data/output/chinext_growth_momentum_quality
 ```
 
 ## Downstream use

README.zh-CN.md

@@ -15,6 +15,7 @@
 | Profile | 名称 | 契约 | Builder |
 | --- | --- | --- | --- |
 | `cn_dividend_quality_snapshot` | CN Dividend Quality Snapshot | `cn_dividend_quality_snapshot.factor_snapshot.v1` | `cneq-build-dividend-quality-snapshot` |
+| `cn_chinext_growth_momentum_quality_snapshot` | CN ChiNext Growth Momentum Quality Snapshot | `cn_chinext_growth_momentum_quality_snapshot.factor_snapshot.v1` | `cneq-build-chinext-growth-momentum-quality-snapshot` |
 
 ## 快速开始
 
@@ -27,6 +28,7 @@ python -m pytest -q
 
 ```bash
 PYTHONPATH=src python scripts/build_dividend_quality_sample.py
+PYTHONPATH=src python scripts/build_chinext_growth_momentum_quality_sample.py
 ```
 
 ## 下游使用

docs/artifact_contract.md

@@ -36,3 +36,47 @@ Contract version: `cn_dividend_quality_snapshot.factor_snapshot.v1`
 - `cn_dividend_quality_snapshot_factor_snapshot_latest.csv.manifest.json`
 - `cn_dividend_quality_snapshot_ranking_latest.csv`
 - `release_status_summary.json`
+
+---
+
+# Artifact contract: cn_chinext_growth_momentum_quality_snapshot
+
+Contract version: `cn_chinext_growth_momentum_quality_snapshot.factor_snapshot.v1`
+
+## Required columns
+
+| Column | Description |
+| --- | --- |
+| `symbol` | 6-digit ChiNext stock code, optional `.SZ` suffix in upstream feeds |
+| `sector` | Industry sector label |
+| `close_cny` | Latest close in CNY |
+| `adv20_cny` | 20-day average daily value traded in CNY |
+| `market_cap_cny` | Market capitalization in CNY |
+| `revenue_yoy` | Revenue year-over-year growth |
+| `profit_yoy` | Profit year-over-year growth |
+| `revenue_acceleration_2q` | Two-quarter revenue acceleration |
+| `roe_ttm` | Return on equity TTM |
+| `roe_stability_3y` | ROE stability score (0-1) |
+| `gross_margin_stability_3y` | Gross margin stability score (0-1) |
+| `mom_12_1` | 12-1 month momentum |
+| `mom_6_1` | 6-1 month momentum |
+| `sma200_gap` | Close vs 200-day moving average gap |
+| `realized_vol_126` | 126-day realized volatility |
+| `earnings_positive` | Whether latest earnings are positive |
+| `suspension_days_63` | Suspension days in last 63 sessions |
+| `is_st` | ST/*ST flag |
+| `list_days` | Listed trading days |
+
+## A-share constraints
+
+- Exclude `is_st=True` rows before ranking.
+- Prefer `list_days >= 252` and `market_cap_cny >= 2e9` for production lineage.
+- Treat this as a growth sleeve: use stronger liquidity and crowding checks than the dividend profile.
+- Snapshot producer must be point-in-time safe; no future financial fields.
+
+## Artifact pack
+
+- `cn_chinext_growth_momentum_quality_snapshot_factor_snapshot_latest.csv`
+- `cn_chinext_growth_momentum_quality_snapshot_factor_snapshot_latest.csv.manifest.json`
+- `cn_chinext_growth_momentum_quality_snapshot_ranking_latest.csv`
+- `release_status_summary.json`

examples/chinext_growth_momentum_quality/factor_snapshot.sample.csv

@@ -0,0 +1,5 @@
+symbol,sector,close_cny,adv20_cny,market_cap_cny,revenue_yoy,profit_yoy,revenue_acceleration_2q,roe_ttm,roe_stability_3y,gross_margin_stability_3y,mom_12_1,mom_6_1,sma200_gap,realized_vol_126,earnings_positive,suspension_days_63,is_st,list_days
+300001,software,28.0,80000000.0,25000000000.0,0.32,0.28,0.10,0.18,0.75,0.70,0.24,0.16,0.12,0.28,True,0,False,1200
+300002,biotech,44.0,60000000.0,18000000000.0,0.24,0.21,0.06,0.15,0.68,0.66,0.18,0.12,0.08,0.24,True,0,False,900
+300003,new_energy,16.0,55000000.0,14000000000.0,0.04,-0.02,-0.01,0.08,0.50,0.48,0.05,0.03,0.01,0.20,True,0,False,800
+159915,etf,3.5,1000000000.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.12,True,0,False,3000

pyproject.toml

@@ -10,7 +10,7 @@ readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
   "pandas>=2.0",
-  "cn-equity-strategies @ git+https://github.com/QuantStrategyLab/CnEquityStrategies.git@8dfadcf8a4dc6cc516f27a4013248474603d8ce2",
+  "cn-equity-strategies @ git+https://github.com/QuantStrategyLab/CnEquityStrategies.git@1bced052d515373e88620dae48499a4ad44c10f7",
 ]
 
 [project.optional-dependencies]
@@ -19,6 +19,7 @@ public-data = ["akshare>=1.14"]
 
 [project.scripts]
 cneq-build-dividend-quality-snapshot = "cn_equity_snapshot_pipelines.dividend_quality:main"
+cneq-build-chinext-growth-momentum-quality-snapshot = "cn_equity_snapshot_pipelines.chinext_growth_momentum_quality:main"
 cneq-stage-akshare-dividend-quality = "cn_equity_snapshot_pipelines.akshare_staging:main"
 cneq-stage-akshare-market-history = "cn_equity_snapshot_pipelines.akshare_market_history:main"
 

scripts/build_chinext_growth_momentum_quality_sample.py

@@ -0,0 +1,23 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+from pathlib import Path
+
+from cn_equity_snapshot_pipelines.chinext_growth_momentum_quality import build_and_write_snapshot
+
+ROOT = Path(__file__).resolve().parents[1]
+SAMPLE = ROOT / "examples" / "chinext_growth_momentum_quality" / "factor_snapshot.sample.csv"
+OUTPUT = ROOT / "data" / "output" / "chinext_growth_momentum_quality"
+
+
+def main() -> int:
+    result = build_and_write_snapshot(
+        factor_snapshot_path=SAMPLE,
+        output_dir=OUTPUT,
+    )
+    print(f"built snapshot rows={len(result.snapshot)} ranking rows={len(result.ranking)}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/cn_equity_snapshot_pipelines/chinext_growth_momentum_quality.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+import argparse
+from datetime import date, datetime, timezone
+from pathlib import Path
+
+import pandas as pd
+
+from cn_equity_strategies.strategies.cn_chinext_growth_momentum_quality_snapshot import (
+    compute_signals,
+    score_candidates,
+)
+
+from .artifacts import SnapshotBuildResult, write_release_status_summary, write_snapshot_manifest
+from .contracts import CN_CHINEXT_GROWTH_MOMENTUM_QUALITY_SNAPSHOT_PROFILE, get_profile_contract
+
+
+def _read_csv(path: str | Path) -> pd.DataFrame:
+    return pd.read_csv(Path(path))
+
+
+def _ensure_snapshot_as_of(snapshot: pd.DataFrame, *, as_of: str | date | None = None) -> pd.DataFrame:
+    frame = snapshot.copy()
+    if "as_of" in frame.columns or "snapshot_date" in frame.columns:
+        return frame
+    stamp = as_of or datetime.now(timezone.utc).date().isoformat()
+    frame.insert(0, "as_of", str(stamp))
+    return frame
+
+
+def build_and_write_snapshot(
+    *,
+    factor_snapshot_path: str | Path,
+    output_dir: str | Path,
+    min_adv20_cny: float = 0.0,
+    min_market_cap_cny: float = 0.0,
+    as_of: str | date | None = None,
+) -> SnapshotBuildResult:
+    contract = get_profile_contract(CN_CHINEXT_GROWTH_MOMENTUM_QUALITY_SNAPSHOT_PROFILE)
+    artifact_paths = contract.artifact_paths(output_dir)
+    snapshot = _ensure_snapshot_as_of(_read_csv(factor_snapshot_path), as_of=as_of)
+    ranking = score_candidates(
+        snapshot,
+        min_adv20_cny=float(min_adv20_cny),
+        min_market_cap_cny=float(min_market_cap_cny),
+    )
+
+    for path in artifact_paths.values():
+        path.parent.mkdir(parents=True, exist_ok=True)
+    snapshot.to_csv(artifact_paths["snapshot"], index=False)
+    ranking.to_csv(artifact_paths["ranking"], index=False)
+    write_snapshot_manifest(
+        contract=contract,
+        snapshot_path=artifact_paths["snapshot"],
+        snapshot=snapshot,
+        manifest_path=artifact_paths["manifest"],
+    )
+    weights, signal_description, _is_hard_defense, status_description, diagnostics = compute_signals(
+        snapshot,
+        current_holdings=set(),
+        min_adv20_cny=float(min_adv20_cny),
+        min_market_cap_cny=float(min_market_cap_cny),
+    )
+    write_release_status_summary(
+        contract=contract,
+        snapshot_path=artifact_paths["snapshot"],
+        manifest_path=artifact_paths["manifest"],
+        ranking_path=artifact_paths["ranking"],
+        summary_path=artifact_paths["release_summary"],
+        snapshot=snapshot,
+        signal_description=signal_description,
+        status_description=status_description,
+        diagnostics={**diagnostics, "target_weights": weights},
+    )
+    return SnapshotBuildResult(
+        snapshot=snapshot,
+        ranking=ranking,
+        artifact_paths=artifact_paths,
+        signal_description=signal_description,
+        status_description=status_description,
+        diagnostics=dict(diagnostics),
+    )
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--factor-snapshot",
+        required=True,
+        help="CSV with cn_chinext_growth_momentum_quality_snapshot factor columns",
+    )
+    parser.add_argument("--output-dir", default="data/output/chinext_growth_momentum_quality")
+    parser.add_argument("--min-adv20-cny", type=float, default=0.0)
+    parser.add_argument("--min-market-cap-cny", type=float, default=0.0)
+    parser.add_argument("--as-of", default=None, help="Snapshot as_of date (YYYY-MM-DD). Defaults to UTC today.")
+    args = parser.parse_args(argv)
+
+    result = build_and_write_snapshot(
+        factor_snapshot_path=args.factor_snapshot,
+        output_dir=args.output_dir,
+        min_adv20_cny=args.min_adv20_cny,
+        min_market_cap_cny=args.min_market_cap_cny,
+        as_of=args.as_of,
+    )
+    print(f"snapshot={result.artifact_paths['snapshot']}")
+    print(f"manifest={result.artifact_paths['manifest']}")
+    print(f"ranking={result.artifact_paths['ranking']}")
+    print(f"release_summary={result.artifact_paths['release_summary']}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/cn_equity_snapshot_pipelines/contracts.py

@@ -5,6 +5,7 @@
 
 SOURCE_PROJECT = "CnEquitySnapshotPipelines"
 CN_DIVIDEND_QUALITY_SNAPSHOT_PROFILE = "cn_dividend_quality_snapshot"
+CN_CHINEXT_GROWTH_MOMENTUM_QUALITY_SNAPSHOT_PROFILE = "cn_chinext_growth_momentum_quality_snapshot"
 
 # Index metadata shared by contracts and index_membership.
 INDEX_METADATA: dict[str, dict[str, str]] = {
@@ -51,6 +52,19 @@ def artifact_paths(self, output_dir: str | Path) -> dict[str, Path]:
         ),
         manifest_required_by_runtime=True,
     ),
+    CN_CHINEXT_GROWTH_MOMENTUM_QUALITY_SNAPSHOT_PROFILE: SnapshotProfileContract(
+        profile=CN_CHINEXT_GROWTH_MOMENTUM_QUALITY_SNAPSHOT_PROFILE,
+        display_name="CN ChiNext Growth Momentum Quality Snapshot",
+        contract_version="cn_chinext_growth_momentum_quality_snapshot.factor_snapshot.v1",
+        snapshot_filename="cn_chinext_growth_momentum_quality_snapshot_factor_snapshot_latest.csv",
+        manifest_filename="cn_chinext_growth_momentum_quality_snapshot_factor_snapshot_latest.csv.manifest.json",
+        ranking_filename="cn_chinext_growth_momentum_quality_snapshot_ranking_latest.csv",
+        legacy_aliases=(),
+        neutral_gcs_prefix_hint=(
+            "gs://qsl-runtime-logs-shared/strategy-artifacts/cn_equity/cn_chinext_growth_momentum_quality_snapshot"
+        ),
+        manifest_required_by_runtime=False,
+    ),
 }
 
 # Index membership timeline profiles (data-pipeline contracts, not runtime strategies).

tests/test_chinext_growth_momentum_quality.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from cn_equity_snapshot_pipelines.chinext_growth_momentum_quality import build_and_write_snapshot
+
+
+def test_build_and_write_chinext_growth_momentum_quality_snapshot(tmp_path: Path) -> None:
+    sample_path = (
+        Path(__file__).resolve().parents[1]
+        / "examples"
+        / "chinext_growth_momentum_quality"
+        / "factor_snapshot.sample.csv"
+    )
+
+    result = build_and_write_snapshot(factor_snapshot_path=sample_path, output_dir=tmp_path)
+
+    assert result.artifact_paths["snapshot"].exists()
+    assert result.artifact_paths["manifest"].exists()
+    assert result.artifact_paths["ranking"].exists()
+    assert result.artifact_paths["release_summary"].exists()
+    assert result.diagnostics["snapshot_contract_version"] == (
+        "cn_chinext_growth_momentum_quality_snapshot.factor_snapshot.v1"
+    )

tests/test_contracts.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from cn_equity_snapshot_pipelines.contracts import (
+    CN_CHINEXT_GROWTH_MOMENTUM_QUALITY_SNAPSHOT_PROFILE,
     CN_DIVIDEND_QUALITY_SNAPSHOT_PROFILE,
     get_profile_contract,
     list_profile_contracts,
@@ -12,10 +13,11 @@ def test_contract_surface_includes_strategy_and_index_membership():
     profiles = [c.profile for c in contracts]
 
     assert CN_DIVIDEND_QUALITY_SNAPSHOT_PROFILE in profiles
+    assert CN_CHINEXT_GROWTH_MOMENTUM_QUALITY_SNAPSHOT_PROFILE in profiles
     assert "cn_csi500_membership" in profiles
     assert "cn_csi300_membership" in profiles
     assert "cn_csi1000_membership" in profiles
-    assert len(profiles) == 4
+    assert len(profiles) == 5
 
     contract = get_profile_contract("cn-dividend-quality-snapshot")
     assert contract.contract_version == "cn_dividend_quality_snapshot.factor_snapshot.v1"
@@ -25,3 +27,7 @@ def test_contract_surface_includes_strategy_and_index_membership():
     csi500 = get_profile_contract("cn_csi500_membership")
     assert csi500.contract_version == "cn_csi500_membership.timeline.v1"
     assert csi500.manifest_required_by_runtime is False
+
+    chinext = get_profile_contract("cn-chinext-growth-momentum-quality-snapshot")
+    assert chinext.contract_version == "cn_chinext_growth_momentum_quality_snapshot.factor_snapshot.v1"
+    assert chinext.manifest_required_by_runtime is False