feat(sector_radar): enhance sector radar functionality with active moneyflow and detailed metrics

- Introduced ActiveMoneyflowSource to fetch optional active-order flow, enhancing the sector radar's data capabilities. - Updated StockFactRecord and DailyAggregateRecord to include pct_change and active_buy_net_amount_yuan for improved financial insights. - Modified the build process to incorporate active moneyflow data without invalidating main rankings on failure. - Enhanced the HTTP API to return detailed sector history and metrics, including pct_change and active buy metrics for members. - Updated tests to validate the new functionality and ensure data integrity across various scenarios.
2026-09-06 16:06:17 +08:00
parent 68282f5d46
commit 7f93d6b0f5
48 changed files with 4883 additions and 95 deletions
@@ -3,13 +3,14 @@ from dataclasses import replace
 from datetime import UTC, date, datetime, timedelta
 from decimal import Decimal
 from pathlib import Path
+from unittest.mock import patch

 import psycopg
 import pytest
 from alembic import command
 from alembic.config import Config

-from zhixing_server.bootstrap.config import sqlalchemy_database_url
+from zhixing_server.bootstrap.config import Settings, sqlalchemy_database_url
 from zhixing_server.modules.sector_radar.domain.models import (
    MembershipStatus,
    PublicationStatus,
@@ -35,7 +36,12 @@ def prepare_database(database_url: str) -> None:
    config = Config(str(server_root / "alembic.ini"))
    sqlalchemy_url = sqlalchemy_database_url(database_url)
    config.set_main_option("sqlalchemy.url", sqlalchemy_url.replace("%", "%%"))
-    command.upgrade(config, "head")
+    config.config_file_name = None
+    with patch(
+        "zhixing_server.bootstrap.config.get_settings",
+        return_value=Settings(database_url=database_url),
+    ):
+        command.upgrade(config, "head")


@pytest.mark.integration
@@ -111,12 +117,22 @@ def test_postgres_sector_radar_revisions_and_last_good() -> None:
                        status=StockFactStatus.AVAILABLE,
                        turnover_yuan=Decimal("1000"),
                        net_amount_yuan=Decimal("100"),
+                        pct_change=Decimal("1.25"),
+                        active_buy_net_amount_yuan=Decimal("-25000"),
                    ),
                )
            ).inserted
            == 1
        )

+        with psycopg.connect(database_url) as connection:
+            detail_fact = connection.execute(
+                "SELECT pct_change, active_buy_net_amount_yuan "
+                "FROM sector_radar_stock_fact WHERE fact_revision = %s",
+                (fact_revision,),
+            ).fetchone()
+            assert detail_fact == (Decimal("1.25"), Decimal("-25000"))
+
        running = RadarPublication(
            publication_id=publication_ids[0],
            target_trade_date=TARGET_DATE,
@@ -639,3 +639,229 @@ def test_history_uses_latest_successful_input_revision_for_a_date() -> None:
    assert second.status == "success"
    assert len(history) == 2
    assert all(item.net_amount_yuan == Decimal(300_000) for item in history)
+
+
+def test_detail_history_is_thirty_sessions_past_only_and_latest_revision() -> None:
+    from zhixing_server.modules.sector_radar.application.details import ReadRadarDetails
+
+    repository = InMemorySectorRadarRepository()
+    end = TARGET_DATE + timedelta(days=33)
+    summary = BuildSectorRadar(FakeRadarSource(), repository, now_fn=lambda: NOW).execute(
+        BuildSectorRadarCommand(start_date=TARGET_DATE, end_date=end)
+    )
+    assert summary.status == "success"
+    target = end - timedelta(days=1)
+    replacement = BuildSectorRadar(
+        FakeRadarSource(net_scale=Decimal(2)),
+        repository,
+        now_fn=lambda: NOW + timedelta(hours=1),
+    ).execute(BuildSectorRadarCommand(trade_date=target))
+    reader = ReadRadarDetails(repository)
+    history = reader.history(target, SectorType.CONCEPT, "BK0001.DC")
+    assert len(history.points) == 30
+    assert history.points[0].trade_date == target - timedelta(days=29)
+    assert history.points[-1].trade_date == target
+    assert len({point.trade_date for point in history.points}) == 30
+    assert history.points[-1].publication_id == replacement.outcomes[0].publication_id
+    assert history.points[-1].amount.metric_value == Decimal("0.003")
+    assert history.available_days == 30
+    assert (
+        reader.history(TARGET_DATE - timedelta(days=1), SectorType.CONCEPT, "BK0001.DC").status
+        == "no_data"
+    )
+
+    old = repository.get_successful_publication(target - timedelta(days=1))
+    assert old is not None
+    repository.publications[old.publication_id] = replace(old, source_version="incompatible-v2")
+    isolated = reader.history(target, SectorType.CONCEPT, "BK0001.DC")
+    assert isolated.points[-2].amount.missing
+    assert isolated.available_days == 29
+
+
+def test_optional_moneyflow_failure_keeps_main_rankings_and_nullable_details() -> None:
+    from zhixing_server.modules.sector_radar.application.details import ReadRadarDetails
+    from zhixing_server.modules.sector_radar.domain.source import MoneyflowRow
+
+    class UnavailableActiveSource(FakeRadarSource):
+        def fetch_moneyflow(self, trade_date: date) -> SourceResult[MoneyflowRow]:
+            raise SourceContractError("optional provider unavailable")
+
+    repository = InMemorySectorRadarRepository()
+    summary = BuildSectorRadar(UnavailableActiveSource(), repository, now_fn=lambda: NOW).execute(
+        BuildSectorRadarCommand(trade_date=TARGET_DATE)
+    )
+    assert summary.status == "success"
+    detail = ReadRadarDetails(repository).detail(TARGET_DATE, SectorType.CONCEPT, "BK0001.DC")
+    assert detail.pct_change == Decimal(1)
+    assert len(detail.members) == 5
+    assert all(member.active_buy_net_amount_yuan is None for member in detail.members)
+    assert detail.members[0].net_amount_yuan == Decimal(10000)
+    assert detail.members[0].pct_change == Decimal(0)
+    assert detail.leaders["active_buy_net_amount_yuan"].top == ()
+    assert detail.summary["amount"].metric_value == Decimal("0.0015")
+
+
+def test_detail_jaccard_uses_same_day_current_listed_members_and_independent_values() -> None:
+    from zhixing_server.modules.sector_radar.application.details import ReadRadarDetails
+    from zhixing_server.modules.sector_radar.domain.source import MoneyflowRow
+
+    class DetailSource(FakeRadarSource):
+        def fetch_sector_members(
+            self, trade_date: date, sector_codes: Sequence[str]
+        ) -> SourceResult[SectorMemberRow]:
+            rows = tuple(
+                SectorMemberRow(trade_date, code, f"00000{index}.SZ", f"股票{index}")
+                for code in sector_codes
+                for index in ((1, 2, 6) if code == "BK0001.DC" else (2, 3))
+            )
+            return self._result("dc_member", trade_date, rows)
+
+        def fetch_moneyflow(self, trade_date: date) -> SourceResult[MoneyflowRow]:
+            raw = (
+                {
+                    "trade_date": trade_date.isoformat(),
+                    "ts_code": "000001.SZ",
+                    "net_mf_amount": "-2.5",
+                },
+            )
+            snapshot = build_source_snapshot(
+                api_name="moneyflow",
+                params={},
+                rows=raw,
+                target_trade_date=trade_date,
+                observed_at=NOW,
+            )
+            return SourceResult((snapshot,), tuple(MoneyflowRow.from_mapping(row) for row in raw))
+
+    repository = InMemorySectorRadarRepository()
+    result = BuildSectorRadar(DetailSource(), repository, now_fn=lambda: NOW).execute(
+        BuildSectorRadarCommand(trade_date=TARGET_DATE)
+    )
+    assert result.status == "success"
+    reader = ReadRadarDetails(repository)
+    detail = reader.detail(TARGET_DATE, SectorType.CONCEPT, "BK0001.DC")
+    assert [member.ts_code for member in detail.members] == ["000001.SZ", "000002.SZ"]
+    assert detail.members[0].active_buy_net_amount_yuan == Decimal(-25000)
+    assert detail.members[0].net_amount_yuan == Decimal(10000)
+    assert detail.members[1].active_buy_net_amount_yuan is None
+    assert len(detail.leaders["active_buy_net_amount_yuan"].top) == 1
+    assert detail.similar_sectors[0].intersection_count == 1
+    assert detail.similar_sectors[0].union_count == 3
+    assert detail.similar_sectors[0].overlap_ratio == Decimal(1) / Decimal(3)
+    assert detail.similar_sectors[0].sector_type is SectorType.INDUSTRY
+    # A later build with different membership cannot alter the older detail.
+    BuildSectorRadar(FakeRadarSource(), repository, now_fn=lambda: NOW).execute(
+        BuildSectorRadarCommand(trade_date=TARGET_DATE + timedelta(days=1))
+    )
+    assert reader.detail(TARGET_DATE, SectorType.CONCEPT, "BK0001.DC") == detail
+
+
+def test_detail_history_and_ranking_extras_http_use_the_same_publication() -> None:
+    from fastapi.testclient import TestClient
+
+    from zhixing_server.bootstrap.app import create_app
+    from zhixing_server.modules.sector_radar.application.read import ReadSectorRadar
+    from zhixing_server.modules.sector_radar.presentation.http import get_sector_radar_reader
+
+    repository = InMemorySectorRadarRepository()
+    result = BuildSectorRadar(FakeRadarSource(), repository, now_fn=lambda: NOW).execute(
+        BuildSectorRadarCommand(trade_date=TARGET_DATE)
+    )
+    app = create_app()
+    app.dependency_overrides[get_sector_radar_reader] = lambda: ReadSectorRadar(repository)
+    with TestClient(app) as client:
+        params = {"trade_date": TARGET_DATE.isoformat()}
+        base = "/api/v1/sector-radar/sectors/concept/BK0001.DC"
+        history = client.get(base + "/history", params=params)
+        detail = client.get(base + "/detail", params=params)
+        ranking = client.get(
+            "/api/v1/sector-radar/rankings", params={**params, "view": "amount", "side": "top"}
+        )
+        assert history.status_code == detail.status_code == ranking.status_code == 200
+        payload = detail.json()
+        assert payload["publication"]["publication_id"] == result.outcomes[0].publication_id
+        assert payload["history"] == history.json()
+        assert payload["pct_change"] == "1"
+        assert payload["members"][0]["active_buy_net_amount_yuan"] is None
+        row = ranking.json()["rows"][0]
+        assert row["pct_change"] == "1"
+        assert Decimal(row["daily_net_amount_yuan"]) == 150000
+        assert Decimal(row["daily_ratio"]) == Decimal("0.03")
+        assert row["on_list_count"] == row["history_available_days"] == 1
+        assert client.get(base + "/detail").status_code == 422
+        absent = client.get(base + "/detail", params={"trade_date": "2020-01-01"})
+        assert absent.json()["status"] == "no_data"
+        assert absent.json()["members"] == []
+
+
+@pytest.mark.integration
+def test_postgres_detail_migration_and_build_roundtrip(monkeypatch: pytest.MonkeyPatch) -> None:
+    import os
+    from pathlib import Path
+
+    import psycopg
+    from alembic import command
+    from alembic.config import Config
+
+    from zhixing_server.bootstrap.config import sqlalchemy_database_url
+    from zhixing_server.modules.sector_radar.application.details import ReadRadarDetails
+    from zhixing_server.modules.sector_radar.infrastructure.postgres import (
+        PostgresSectorRadarRepository,
+    )
+
+    database_url = os.getenv("ZHIXING_TEST_DATABASE_URL")
+    if not database_url:
+        pytest.skip("set ZHIXING_TEST_DATABASE_URL to run PostgreSQL integration tests")
+    config = Config(str(Path(__file__).parents[3] / "alembic.ini"))
+    config.set_main_option(
+        "sqlalchemy.url", sqlalchemy_database_url(database_url).replace("%", "%%")
+    )
+    # Alembic intentionally reads Settings; bind the explicit test DSN and avoid
+    # fileConfig disabling unrelated test loggers in the same pytest process.
+    config.config_file_name = None
+    with monkeypatch.context() as context:
+        context.setenv("ZHIXING_DATABASE_URL", database_url)
+        from zhixing_server.bootstrap.config import get_settings
+
+        get_settings.cache_clear()
+        try:
+            command.upgrade(config, "head")
+        finally:
+            get_settings.cache_clear()
+    target = date(2098, 12, 1)
+    repository = PostgresSectorRadarRepository(database_url, max_connections=2)
+    try:
+        result = BuildSectorRadar(FakeRadarSource(), repository, now_fn=lambda: NOW).execute(
+            BuildSectorRadarCommand(trade_date=target)
+        )
+        assert result.status in ("success", "unchanged")
+        detail = ReadRadarDetails(repository).detail(target, SectorType.CONCEPT, "BK0001.DC")
+        assert detail.history.status == "success"
+        assert detail.pct_change == Decimal(1)
+        assert len(detail.members) == 5
+        assert detail.members[0].active_buy_net_amount_yuan is None
+        assert detail.summary["amount"].metric_value == Decimal("0.0015")
+        with psycopg.connect(database_url) as connection:
+            assert connection.execute("SELECT version_num FROM alembic_version").fetchone() == (
+                "0009_radar_sector_detail",
+            )
+            row = connection.execute(
+                "SELECT pct_change, leading_code FROM sector_radar_daily_aggregate "
+                "WHERE publication_id = %s AND sector_type = 'concept'",
+                (result.outcomes[0].publication_id,),
+            ).fetchone()
+            assert row == (Decimal(1), "000001.SZ")
+            facts = connection.execute(
+                "SELECT pct_change, active_buy_net_amount_yuan "
+                "FROM sector_radar_stock_fact WHERE trade_date = %s",
+                (target,),
+            ).fetchall()
+            assert facts and all(row == (Decimal(0), None) for row in facts)
+            with pytest.raises(psycopg.errors.CheckViolation), connection.transaction():
+                connection.execute(
+                    "UPDATE sector_radar_stock_fact "
+                    "SET active_buy_net_amount_yuan = 'NaN'::numeric WHERE trade_date = %s",
+                    (target,),
+                )
+    finally:
+        repository.close()
@@ -402,3 +402,40 @@ def test_sector_member_codes_rejects_blank_sector_code() -> None:

    with pytest.raises(ValueError):
        reader.sector_member_codes(TARGET_DATE, "   ")
+
+
+def test_history_appearance_counts_use_each_days_pool_and_metric_version() -> None:
+    from zhixing_server.modules.sector_radar.application.details import ReadRadarDetails
+
+    repository = InMemorySectorRadarRepository()
+    current_rows = _amount_rankings()
+    for offset, size in ((-1, 5), (0, 10), (1, 20)):
+        day = TARGET_DATE + timedelta(days=offset)
+        running = _running(f"pool-{offset}", day)
+        repository.create_publication(running)
+        repository.finish_publication(_finish(running, PublicationStatus.SUCCESS))
+        observations = tuple(
+            replace(row.observation, trade_date=day) for row in current_rows[:size]
+        )
+        repository.save_rankings(
+            tuple(
+                RankingRecord(running.publication_id, row)
+                for row in rank_metric_observations(observations)
+            )
+        )
+    reader = ReadRadarDetails(repository)
+    history = reader.history(TARGET_DATE, SectorType.CONCEPT, "BK0002.DC")
+    assert [point.amount.pool_size for point in history.points] == [5, 10]
+    assert [point.amount.in_top for point in history.points] == [False, True]
+    publication = repository.get_successful_publication(TARGET_DATE)
+    assert publication is not None
+    extra = reader.ranking_extras(publication, (current_rows[1],), RankSide.TOP)["BK0002.DC"]
+    assert extra.on_list_count == 1
+    assert extra.history_available_days == 2
+    old_version = replace(
+        current_rows[1],
+        observation=replace(current_rows[1].observation, metric_version="older-version"),
+    )
+    from zhixing_server.modules.sector_radar.application.details import metric_at
+
+    assert metric_at((old_version,), SectorType.CONCEPT, "BK0002.DC", MetricKind.AMOUNT).missing