feat(sector-radar): add weighted scores and rank-change views

2026-09-21 23:57:55 +08:00
parent 669e89d3c3
commit b9981aa48d
45 changed files with 2468 additions and 320 deletions
@@ -18,14 +18,10 @@ from zhixing_server.shared.request_coordinator import TushareSourceError

 from ..domain.facts import aggregate_sector_snapshot
 from ..domain.metrics import (
-    AmountNetStrategy,
    MetricStrategy,
-    RatioTurnoverStrategy,
-    SwingEqualThreeToTenStrategy,
 )
 from ..domain.models import (
    MembershipStatus,
-    MetricObservation,
    PublicationStatus,
    RadarPublication,
    RankedMetric,
@@ -50,7 +46,6 @@ from ..domain.persistence import (
    StockFactRecord,
 )
 from ..domain.ports import ActiveMoneyflowSource, SectorRadarSource
-from ..domain.ranking import rank_metric_observations, with_rank_changes
 from ..domain.source import (
    DailyRow,
    MoneyflowDcRow,
@@ -65,6 +60,7 @@ from ..domain.source import (
    SuspendRow,
    TradeCalendarRow,
 )
+from .scoring import calculate_rankings, calendar_rank_changes, default_strategies

 BuildOutcomeStatus = Literal["success", "partial", "failed", "locked", "unchanged"]
 SHANGHAI = ZoneInfo("Asia/Shanghai")
@@ -194,14 +190,7 @@ class BuildSectorRadar:
        self.repository = repository
        self.coverage_threshold = coverage_threshold
        self.today = today or self.now_fn().astimezone(SHANGHAI).date()
-        self.strategies = tuple(
-            strategies
-            or (
-                AmountNetStrategy(),
-                RatioTurnoverStrategy(),
-                SwingEqualThreeToTenStrategy(),
-            )
-        )
+        self.strategies = tuple(strategies) if strategies is not None else default_strategies()

    def execute(self, command: BuildSectorRadarCommand | None = None) -> BuildSummary:
        """Build each selected trade date sequentially for deterministic history."""
@@ -314,7 +303,13 @@ class BuildSectorRadar:
            publication_created = True
            reusable = self._reusable_sources(target.retry_publication_id)
            collected = self._collect(target.trade_date, publication_id, reusable)
-            input_hash = self._input_hash(collected.snapshots)
+            previous_publications = tuple(
+                item
+                for item in self.repository.load_history_publications(target.trade_date)
+                if item.target_trade_date < target.trade_date
+                and item.source_version == self.source_version
+            )[:9]
+            input_hash = self._input_hash(collected.snapshots, previous_publications)
            existing = self.repository.find_reusable_publication(target.trade_date, input_hash)
            if existing is not None:
                self.repository.discard_running_publication(publication_id)
@@ -339,7 +334,9 @@ class BuildSectorRadar:
                input_hash=input_hash,
            )
            aggregates = self._aggregate(collected)
-            rankings = self._rank(target.trade_date, aggregates)
+            rankings = self._rank(
+                target.trade_date, aggregates, collected.trading_dates, previous_publications
+            )
            coverage = self._coverage(collected.stock_facts)
            membership_complete = all(
                item.status is MembershipStatus.AVAILABLE for item in collected.memberships
@@ -638,6 +635,7 @@ class BuildSectorRadar:
        )
        return _CollectedInputs(
            target_trade_date=target,
+            trading_dates=tuple(sorted({row.cal_date for row in calendar.rows if row.is_open})),
            indices=concepts.rows + industries.rows,
            snapshots=snapshots,
            membership_snapshots=members.snapshots,
@@ -692,24 +690,29 @@ class BuildSectorRadar:
        return aggregates

    def _rank(
-        self, target: date, aggregates: Sequence[SectorDailyAggregate]
+        self,
+        target: date,
+        aggregates: Sequence[SectorDailyAggregate],
+        trading_dates: Sequence[date],
+        previous_publications: Sequence[RadarPublication],
    ) -> tuple[RankedMetric, ...]:
-        history = tuple(self.repository.load_daily_aggregate_history(target, limit_dates=9))
-        observations: list[MetricObservation] = []
-        for current in aggregates:
-            sector_history = tuple(
-                item
-                for item in history
-                if (item.sector_type, item.sector_code)
-                == (current.sector_type, current.sector_code)
-            ) + (current,)
-            observations.extend(
-                strategy.evaluate(sector_history, target) for strategy in self.strategies
+        history = tuple(
+            record.aggregate
+            for record in self.repository.load_publication_aggregates(
+                tuple(item.publication_id for item in previous_publications)
            )
-        current_rankings = rank_metric_observations(observations)
-        previous = self.repository.load_previous_rankings(target, limit_dates=5)
-        history_by_days = {days: rankings for days, (_, rankings) in enumerate(previous, start=1)}
-        return with_rank_changes(current_rankings, history_by_days)
+        )
+        current_rankings = calculate_rankings(
+            target, aggregates, history, trading_dates, self.strategies
+        )
+        dates_by_id = {
+            item.publication_id: item.target_trade_date for item in previous_publications[:5]
+        }
+        previous = {
+            dates_by_id[key]: rows
+            for key, rows in self.repository.load_publication_rankings(tuple(dates_by_id))
+        }
+        return calendar_rank_changes(current_rankings, previous, trading_dates, target)

    @staticmethod
    def _coverage(stock_facts: Sequence[StockFactRecord]) -> Decimal:
@@ -750,12 +753,15 @@ class BuildSectorRadar:
            groups.append(PublicationSourceGroup.MONEYFLOW_DC)
        return tuple(groups)

-    def _input_hash(self, snapshots: Sequence[SourceSnapshot]) -> str:
+    def _input_hash(
+        self, snapshots: Sequence[SourceSnapshot], previous: Sequence[RadarPublication] = ()
+    ) -> str:
        payload = json.dumps(
            {
                "snapshot_ids": sorted(snapshot.snapshot_id for snapshot in snapshots),
                "metric_versions": sorted(strategy.metric_version for strategy in self.strategies),
                "normalizer": "zhixing_stock_fact_v2",
+                "history": [(item.publication_id, item.input_hash) for item in previous],
            },
            sort_keys=True,
            separators=(",", ":"),
@@ -785,6 +791,7 @@ class BuildSectorRadar:
@dataclass(frozen=True, slots=True)
 class _CollectedInputs:
    target_trade_date: date
+    trading_dates: tuple[date, ...]
    indices: tuple[SectorIndexRow, ...]
    snapshots: tuple[SourceSnapshot, ...]
    membership_snapshots: tuple[SourceSnapshot, ...]
@@ -7,7 +7,6 @@ from dataclasses import dataclass
 from datetime import date
 from decimal import Decimal

-from ..domain.metrics import AmountNetStrategy, RatioTurnoverStrategy, SwingEqualThreeToTenStrategy
 from ..domain.models import MetricKind, RadarPublication, RankedMetric, RankSide, SectorType
 from ..domain.normalize import is_current_listed_stock
 from ..domain.persistence import HistoricalRanking, PublicationSourceGroup, SectorRadarRepository
@@ -20,12 +19,7 @@ from ..domain.source import (
    StockBasicRow,
    TradeCalendarRow,
 )
-
-_METRIC_VERSIONS = {
-    MetricKind.AMOUNT: AmountNetStrategy.metric_version,
-    MetricKind.RATIO: RatioTurnoverStrategy.metric_version,
-    MetricKind.SWING: SwingEqualThreeToTenStrategy.metric_version,
-}
+from ..domain.weighted import resolve_metric_version


@dataclass(frozen=True, slots=True)
@@ -50,6 +44,7 @@ class HistoryMetric:
    missing: bool = True
    in_top: bool = False
    in_bottom: bool = False
+    weighted_score: Decimal | None = None


@dataclass(frozen=True, slots=True)
@@ -134,7 +129,8 @@ class SectorDetail:
 class _RankHistory:
    """Index one day's selected ranks once, retaining complete versioned pool counts."""

-    def __init__(self, records: Sequence[HistoricalRanking]) -> None:
+    def __init__(self, records: Sequence[HistoricalRanking], versions: Sequence[str] = ()) -> None:
+        self.versions = tuple(versions) or tuple(record.metric_version for record in records)
        self.pools: dict[tuple[SectorType, MetricKind, str], int] = {}
        self.rankings: dict[tuple[SectorType, str, MetricKind, str], RankedMetric] = {}
        self.names: dict[tuple[SectorType, str], str] = {}
@@ -158,7 +154,9 @@ class _RankHistory:
                )

    def metric(self, sector_type: SectorType, sector_code: str, kind: MetricKind) -> HistoryMetric:
-        version = _METRIC_VERSIONS[kind]
+        version = resolve_metric_version(kind, self.versions)
+        if version is None:
+            return HistoryMetric()
        size = self.pools.get((sector_type, kind, version), 0)
        row = self.rankings.get((sector_type, sector_code, kind, version))
        return _metric_from_ranking(row, size)
@@ -187,9 +185,9 @@ class ReadRadarDetails:
        ):
            by_id.setdefault(record.publication_id, []).append(record)
        rows = {
-            day: _RankHistory(by_id.get(item.publication_id, ()))
+            day: _RankHistory(by_id.get(item.publication_id, ()), publication.metric_versions)
            if item.source_version == publication.source_version
-            else _RankHistory(())
+            else _RankHistory((), publication.metric_versions)
            for day, item in publications.items()
        }
        sources = self.repository.load_publication_rows(
@@ -209,7 +207,11 @@ class ReadRadarDetails:
                | set(publications)
            )[-30:]
        )
-        return publications, {day: rows.get(day, _RankHistory(())) for day in dates}, dates
+        return (
+            publications,
+            {day: rows.get(day, _RankHistory((), publication.metric_versions)) for day in dates},
+            dates,
+        )

    def ranking_extras(
        self, publication: RadarPublication, rankings: Sequence[RankedMetric], side: RankSide
@@ -438,12 +440,13 @@ def metric_at(
    rows: Sequence[RankedMetric], sector_type: SectorType, sector_code: str, kind: MetricKind
 ) -> HistoryMetric:
    """Select compatible rank values; pool thresholds use each day's actual percentile."""
+    version = resolve_metric_version(kind, (row.observation.metric_version for row in rows))
    pool = [
        row
        for row in rows
        if row.observation.sector_type is sector_type
        and row.observation.metric_kind is kind
-        and row.observation.metric_version == _METRIC_VERSIONS[kind]
+        and row.observation.metric_version == version
    ]
    size = sum(row.rank_position is not None for row in pool)
    row = next((row for row in pool if row.observation.sector_code == sector_code), None)
@@ -463,6 +466,7 @@ def _metric_from_ranking(row: RankedMetric | None, size: int) -> HistoryMetric:
        row.rank_position is None,
        percentile is not None and percentile >= 90,
        percentile is not None and percentile <= 10,
+        row.observation.weighted_score,
    )


@@ -27,7 +27,13 @@ from ..domain.persistence import (
    StockMembershipEntry,
 )
 from ..domain.ranking import select_percentile_side, select_rank_change_side
+from ..domain.weighted import (
+    RatioWeightedStrategy,
+    SwingWeightedStrategy,
+    resolve_metric_version,
+)
 from .details import RankingExtras, ReadRadarDetails
+from .scoring import calendar_rank_changes, comparison_dates, publication_trading_dates

 ReadStatus = Literal["success", "no_data"]

@@ -62,7 +68,7 @@ class RadarQuery:
    trade_date: date | None = None
    sector_type: SectorType = SectorType.CONCEPT
    view: RadarView = RadarView.AMOUNT
-    rank_change_metric: MetricKind = MetricKind.AMOUNT
+    rank_change_metric: MetricKind = MetricKind.SWING
    rank_change_days: int = 1
    side: RankSide = RankSide.ALL
    search: str | None = None
@@ -108,6 +114,10 @@ class RankingPage:
    rows: tuple[RankedMetric, ...]
    total: int
    extras: dict[str, RankingExtras] = field(default_factory=lambda: dict[str, RankingExtras]())
+    rank_change_values: dict[str, dict[MetricKind, int | None]] = field(
+        default_factory=lambda: dict[str, dict[MetricKind, int | None]]()
+    )
+    comparison_trade_date: date | None = None


@dataclass(frozen=True, slots=True)
@@ -175,27 +185,54 @@ class SectorMembersSnapshot:


 _METRIC_DEFINITIONS = {
-    MetricKind.AMOUNT: RadarMetricDefinition(
+    AmountNetStrategy.metric_version: RadarMetricDefinition(
        metric_kind=MetricKind.AMOUNT,
        metric_version=AmountNetStrategy.metric_version,
        label="主力净流入（知行独立实现）",
        unit=MetricUnit.CNY_100M,
    ),
-    MetricKind.RATIO: RadarMetricDefinition(
+    RatioTurnoverStrategy.metric_version: RadarMetricDefinition(
        metric_kind=MetricKind.RATIO,
        metric_version=RatioTurnoverStrategy.metric_version,
        label="主力净流入/成交额（知行独立实现）",
        unit=MetricUnit.RATIO,
    ),
-    MetricKind.SWING: RadarMetricDefinition(
+    SwingEqualThreeToTenStrategy.metric_version: RadarMetricDefinition(
        metric_kind=MetricKind.SWING,
        metric_version=SwingEqualThreeToTenStrategy.metric_version,
        label="3—10 日等权资金率（知行独立实现）",
        unit=MetricUnit.RATIO,
    ),
+    RatioWeightedStrategy.metric_version: RadarMetricDefinition(
+        metric_kind=MetricKind.RATIO,
+        metric_version=RatioWeightedStrategy.metric_version,
+        label="单日流入率",
+        unit=MetricUnit.RATIO,
+        disclaimer="基于知行数据的资金排名与成交额加权评分",
+    ),
+    SwingWeightedStrategy.metric_version: RadarMetricDefinition(
+        metric_kind=MetricKind.SWING,
+        metric_version=SwingWeightedStrategy.metric_version,
+        label="波段流入率",
+        unit=MetricUnit.RATIO,
+        disclaimer="3 日与 10 日资金率排名各占 50%，再按成交额加权",
+    ),
 }


+def metric_definition(
+    kind: MetricKind, publication: RadarPublication | None
+) -> RadarMetricDefinition:
+    """Resolve a publication's actual algorithm instead of hiding legacy rows."""
+    current = {
+        MetricKind.AMOUNT: AmountNetStrategy.metric_version,
+        MetricKind.RATIO: RatioWeightedStrategy.metric_version,
+        MetricKind.SWING: SwingWeightedStrategy.metric_version,
+    }[kind]
+    version = resolve_metric_version(kind, publication.metric_versions) if publication else None
+    return _METRIC_DEFINITIONS[version or current]
+
+
 class ReadSectorRadar:
    """Hide last-good selection, ranking filters, search, and pagination."""

@@ -219,18 +256,45 @@ class ReadSectorRadar:
            if query.view is RadarView.RANK_CHANGE
            else MetricKind(query.view.value)
        )
-        definition = _METRIC_DEFINITIONS[metric_kind]
        publication = (
            self.repository.get_successful_publication(query.trade_date)
            if query.trade_date is not None
            else self.repository.get_last_good_publication()
        )
+        definition = metric_definition(metric_kind, publication)
        if publication is None:
            return RankingPage("no_data", query, None, definition, (), 0)

+        all_rows = tuple(self.repository.load_rankings(publication.publication_id))
+        compared_date = None
+        if query.view is RadarView.RANK_CHANGE:
+            calendar = publication_trading_dates(
+                self.repository, publication.publication_id, publication.target_trade_date
+            )
+            previous_publications = tuple(
+                item
+                for item in self.repository.load_history_publications(publication.target_trade_date)
+                if item.target_trade_date < publication.target_trade_date
+                and item.source_version == publication.source_version
+            )[:5]
+            dates_by_id = {
+                item.publication_id: item.target_trade_date for item in previous_publications
+            }
+            previous = {
+                dates_by_id[publication_id]: rows
+                for publication_id, rows in self.repository.load_publication_rankings(
+                    tuple(dates_by_id)
+                )
+            }
+            all_rows = calendar_rank_changes(
+                all_rows, previous, calendar, publication.target_trade_date
+            )
+            compared_date = comparison_dates(calendar, publication.target_trade_date)[
+                query.rank_change_days
+            ]
        metric_rows = tuple(
            row
-            for row in self.repository.load_rankings(publication.publication_id)
+            for row in all_rows
            if row.observation.sector_type is query.sector_type
            and row.observation.metric_kind is metric_kind
            and row.observation.metric_version == definition.metric_version
@@ -276,18 +340,32 @@ class ReadSectorRadar:
            or search in row.observation.sector_name.casefold()
        )
        start = (query.page - 1) * query.page_size
+        page_rows = searched[start : start + query.page_size]
+        selected_codes = {row.observation.sector_code for row in page_rows}
+        changes: dict[str, dict[MetricKind, int | None]] = {}
+        for row in all_rows:
+            observation = row.observation
+            if (
+                observation.sector_type is query.sector_type
+                and observation.sector_code in selected_codes
+                and observation.metric_version
+                == resolve_metric_version(observation.metric_kind, publication.metric_versions)
+            ):
+                changes.setdefault(observation.sector_code, {})[observation.metric_kind] = (
+                    row.rank_change(query.rank_change_days)
+                )
        return RankingPage(
            status="success",
            query=query,
            publication=publication,
            definition=definition,
-            rows=searched[start : start + query.page_size],
+            rows=page_rows,
            total=len(searched),
            extras=ReadRadarDetails(self.repository).ranking_extras(
-                publication, searched[start : start + query.page_size], query.side
-            )
-            if query.view in {RadarView.AMOUNT, RadarView.RATIO}
-            else {},
+                publication, page_rows, query.side
+            ),
+            rank_change_values=changes,
+            comparison_trade_date=compared_date,
        )

    def stock_membership(self, query: StockSectorQuery) -> StockSectorMembership:
@@ -0,0 +1,268 @@
+"""Rescore immutable local inputs without refetching or altering source history."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from collections.abc import Callable, Sequence
+from dataclasses import replace
+from datetime import UTC, date, datetime
+from decimal import Decimal
+from uuid import uuid4
+
+from ..domain.models import PublicationStatus, RadarPublication, RankedMetric
+from ..domain.persistence import (
+    DailyAggregateRecord,
+    PublicationSourceRecord,
+    RankingRecord,
+    SectorRadarRepository,
+)
+from .build import BuildDateOutcome, BuildSectorRadarCommand, BuildSummary
+from .scoring import (
+    calculate_rankings,
+    calendar_rank_changes,
+    default_strategies,
+    publication_trading_dates,
+)
+
+
+class RecomputeSectorRadar:
+    """Publish a new algorithm revision from pinned, already-normalized facts.
+
+    No source adapter is accepted: current stock membership cannot accidentally
+    replace a historical snapshot. Failed rebuilds leave the old last-good intact.
+    """
+
+    def __init__(
+        self,
+        repository: SectorRadarRepository,
+        now_fn: Callable[[], datetime] = lambda: datetime.now(UTC),
+    ) -> None:
+        self.repository = repository
+        self.now_fn = now_fn
+        self.versions = tuple(strategy.metric_version for strategy in default_strategies())
+
+    def execute(self, command: BuildSectorRadarCommand) -> BuildSummary:
+        """Rescore successful dates oldest first, stopping if one date fails.
+
+        The requested range is inclusive. A single date or no date selects one
+        existing publication. All aggregate inputs are pinned before writes.
+        """
+        if command.retry_publication_id is not None:
+            raise ValueError("offline recompute does not accept a source retry")
+        available = sorted(self.repository.list_successful_dates())
+        if command.trade_date is not None:
+            targets = [command.trade_date] if command.trade_date in available else []
+        elif command.start_date is not None and command.end_date is not None:
+            targets = [day for day in available if command.start_date <= day <= command.end_date]
+        else:
+            targets = available[-1:]
+        if not targets:
+            return BuildSummary(
+                (
+                    BuildDateOutcome(
+                        command.trade_date or command.start_date or self.now_fn().date(),
+                        "failed",
+                        None,
+                        Decimal(0),
+                        0,
+                        0,
+                        "no_local_publication",
+                        "no successful local input publication",
+                    ),
+                )
+            )
+        needed = [day for day in available if day < targets[0]][-9:] + targets
+        pinned: dict[date, RadarPublication] = {}
+        for day in needed:
+            publication = self.repository.get_successful_publication(day)
+            if publication is None:
+                raise ValueError("selected local publication disappeared")
+            pinned[day] = publication
+        inputs = self.repository.load_publication_aggregates(
+            tuple(item.publication_id for item in pinned.values())
+        )
+        by_publication: dict[str, list[DailyAggregateRecord]] = {}
+        for record in inputs:
+            by_publication.setdefault(record.publication_id, []).append(record)
+        existing_rankings = dict(
+            self.repository.load_publication_rankings(
+                tuple(item.publication_id for item in pinned.values())
+            )
+        )
+        ranks_by_day = {
+            day: existing_rankings.get(item.publication_id, ()) for day, item in pinned.items()
+        }
+        outcomes: list[BuildDateOutcome] = []
+        for target in targets:
+            base = pinned[target]
+            current = by_publication.get(base.publication_id, [])
+            prior_dates = [day for day in needed if day < target][-9:]
+            history = [
+                record
+                for day in prior_dates
+                if pinned[day].source_version == base.source_version
+                for record in by_publication.get(pinned[day].publication_id, [])
+            ]
+            previous = {
+                day: ranks_by_day[day]
+                for day in prior_dates
+                if pinned[day].source_version == base.source_version
+            }
+            outcome, rankings = self._rescore(base, current, history, previous)
+            outcomes.append(outcome)
+            if outcome.status in {"failed", "locked"}:
+                break
+            ranks_by_day[target] = rankings
+        return BuildSummary(tuple(outcomes))
+
+    def _rescore(
+        self,
+        base: RadarPublication,
+        current: Sequence[DailyAggregateRecord],
+        history: Sequence[DailyAggregateRecord],
+        previous: dict[date, Sequence[RankedMetric]],
+    ) -> tuple[BuildDateOutcome, Sequence[RankedMetric]]:
+        target = base.target_trade_date
+        pending: RadarPublication | None = None
+        try:
+            with self.repository.advisory_lock(target) as acquired:
+                if not acquired:
+                    return BuildDateOutcome(target, "locked", None, Decimal(0), 0, 0), ()
+                if not current:
+                    raise ValueError("local aggregate input is missing")
+                calendar = publication_trading_dates(self.repository, base.publication_id, target)
+                rankings = calendar_rank_changes(
+                    calculate_rankings(
+                        target,
+                        [item.aggregate for item in current],
+                        [item.aggregate for item in history],
+                        calendar,
+                    ),
+                    previous,
+                    calendar,
+                    target,
+                )
+                sources = tuple(self.repository.load_publication_sources(base.publication_id))
+                digest = self._fingerprint(base, current, history, sources, rankings)
+                reusable = self.repository.find_reusable_publication(target, digest)
+                if reusable is not None and reusable.status is PublicationStatus.SUCCESS:
+                    return BuildDateOutcome(
+                        target,
+                        "unchanged",
+                        reusable.publication_id,
+                        reusable.coverage,
+                        len(current),
+                        len(rankings),
+                    ), rankings
+                started = self.now_fn()
+                self.repository.recover_running_publications(target, finished_at=started)
+                pending = replace(
+                    base,
+                    publication_id=f"radar-{target:%Y%m%d}-rescore-{uuid4().hex[:20]}",
+                    status=PublicationStatus.RUNNING,
+                    metric_versions=self.versions,
+                    input_hash=None,
+                    started_at=started,
+                    finished_at=None,
+                    error_summary=None,
+                )
+                self.repository.create_publication(pending)
+                self.repository.save_publication_sources(
+                    replace(item, publication_id=pending.publication_id, refresh_on_retry=False)
+                    for item in sources
+                )
+                finished = replace(
+                    pending,
+                    status=PublicationStatus.SUCCESS,
+                    input_hash=digest,
+                    finished_at=self.now_fn(),
+                )
+                self.repository.finalize_publication(
+                    finished,
+                    memberships=(),
+                    stock_facts=(),
+                    daily_aggregates=(
+                        replace(item, publication_id=pending.publication_id) for item in current
+                    ),
+                    rankings=(RankingRecord(pending.publication_id, row) for row in rankings),
+                )
+                return BuildDateOutcome(
+                    target,
+                    "success",
+                    pending.publication_id,
+                    base.coverage,
+                    len(current),
+                    len(rankings),
+                ), rankings
+        except (ValueError, RuntimeError, ArithmeticError) as exc:
+            if pending is not None:
+                saved = self.repository.get_publication(pending.publication_id)
+                if saved is not None and saved.status is PublicationStatus.RUNNING:
+                    self.repository.finish_publication(
+                        replace(
+                            pending,
+                            status=PublicationStatus.FAILED,
+                            finished_at=self.now_fn(),
+                            error_summary="offline_recompute_failed",
+                        )
+                    )
+            return BuildDateOutcome(
+                target,
+                "failed",
+                pending.publication_id if pending else None,
+                Decimal(0),
+                0,
+                0,
+                type(exc).__name__,
+                "local radar rescoring failed",
+            ), ()
+
+    def _fingerprint(
+        self,
+        base: RadarPublication,
+        current: Sequence[DailyAggregateRecord],
+        history: Sequence[DailyAggregateRecord],
+        sources: Sequence[PublicationSourceRecord],
+        rankings: Sequence[RankedMetric],
+    ) -> str:
+        """Hash source content, not derived publication IDs, so reruns are idempotent."""
+        payload = {
+            "source_version": base.source_version,
+            "universe_version": base.universe_version,
+            "metric_versions": self.versions,
+            "snapshots": sorted(item.snapshot.snapshot_id for item in sources),
+            "inputs": [
+                (
+                    item.aggregate.trade_date.isoformat(),
+                    item.aggregate.sector_type.value,
+                    item.aggregate.sector_code,
+                    str(item.aggregate.net_amount_yuan),
+                    str(item.aggregate.turnover_yuan),
+                    item.aggregate.member_count,
+                    item.aggregate.valid_sample_count,
+                    str(item.aggregate.membership_coverage),
+                    str(item.aggregate.moneyflow_coverage),
+                    str(item.pct_change),
+                    item.leading_code,
+                )
+                for item in sorted(
+                    (*history, *current),
+                    key=lambda item: (
+                        item.aggregate.trade_date,
+                        item.aggregate.sector_type,
+                        item.aggregate.sector_code,
+                    ),
+                )
+            ],
+            "rank_changes": [
+                (
+                    row.observation.sector_type.value,
+                    row.observation.sector_code,
+                    row.observation.metric_version,
+                    [(change.days, change.value) for change in row.rank_changes],
+                )
+                for row in rankings
+            ],
+        }
+        return hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()
@@ -0,0 +1,98 @@
+"""Shared publication scoring and strict trading-session comparisons."""
+
+from __future__ import annotations
+
+from collections import defaultdict
+from collections.abc import Mapping, Sequence
+from datetime import date
+
+from ..domain.metrics import AmountNetStrategy, MetricStrategy
+from ..domain.models import MetricObservation, RankedMetric, SectorDailyAggregate, SectorType
+from ..domain.persistence import PublicationSourceGroup, SectorRadarRepository
+from ..domain.ranking import rank_metric_observations, with_rank_changes
+from ..domain.source import TradeCalendarRow
+from ..domain.weighted import (
+    FlowFeatures,
+    RatioWeightedStrategy,
+    SwingWeightedStrategy,
+    attach_weighted_scores,
+    calendar_sector_history,
+    flow_features,
+)
+
+
+def default_strategies() -> tuple[MetricStrategy, ...]:
+    """Return versioned strategies shared by online builds and offline rescoring."""
+    return (AmountNetStrategy(), RatioWeightedStrategy(), SwingWeightedStrategy())
+
+
+def publication_trading_dates(
+    repository: SectorRadarRepository,
+    publication_id: str,
+    target: date,
+) -> tuple[date, ...]:
+    """Read the exact publication's calendar without querying an external provider."""
+    sources = repository.load_publication_rows(publication_id, (PublicationSourceGroup.CALENDAR,))
+    return tuple(
+        sorted(
+            {
+                parsed.cal_date
+                for row in sources.get(PublicationSourceGroup.CALENDAR, ())
+                for parsed in (TradeCalendarRow.from_mapping(row),)
+                if parsed.is_open and parsed.cal_date <= target
+            }
+        )
+    )
+
+
+def calculate_rankings(
+    target: date,
+    aggregates: Sequence[SectorDailyAggregate],
+    history: Sequence[SectorDailyAggregate],
+    trading_dates: Sequence[date],
+    strategies: Sequence[MetricStrategy] | None = None,
+) -> tuple[RankedMetric, ...]:
+    """Evaluate raw features, score full pools, then generate authoritative ranks."""
+    histories: defaultdict[tuple[SectorType, str], dict[date, SectorDailyAggregate]] = defaultdict(
+        dict
+    )
+    for row in history:
+        if row.trade_date >= target:
+            continue
+        days = histories[(row.sector_type, row.sector_code)]
+        if row.trade_date in days:
+            raise ValueError("aggregate history must contain unique sector dates")
+        days[row.trade_date] = row
+    observations: list[MetricObservation] = []
+    features: dict[tuple[SectorType, str], FlowFeatures] = {}
+    selected = default_strategies() if strategies is None else strategies
+    for current in aggregates:
+        if current.trade_date != target:
+            raise ValueError("current aggregates must match the target date")
+        key = (current.sector_type, current.sector_code)
+        sector_history = calendar_sector_history(current, histories[key], trading_dates)
+        features[key] = flow_features(sector_history, target)
+        observations.extend(strategy.evaluate(sector_history, target) for strategy in selected)
+    return rank_metric_observations(attach_weighted_scores(observations, features))
+
+
+def comparison_dates(trading_dates: Sequence[date], target: date) -> dict[int, date | None]:
+    """Resolve actual prior trading sessions; missing calendar history stays unknown."""
+    previous = sorted({day for day in trading_dates if day < target}, reverse=True)
+    return {days: previous[days - 1] if len(previous) >= days else None for days in range(1, 6)}
+
+
+def calendar_rank_changes(
+    current: Sequence[RankedMetric],
+    previous: Mapping[date, Sequence[RankedMetric]],
+    trading_dates: Sequence[date],
+    target: date,
+) -> tuple[RankedMetric, ...]:
+    """Compare matching versions on exact calendar dates, without skipping failed days."""
+    return with_rank_changes(
+        current,
+        {
+            days: previous.get(day, ()) if day is not None else ()
+            for days, day in comparison_dates(trading_dates, target).items()
+        },
+    )
@@ -33,22 +33,24 @@ class MetricStrategy(Protocol):
        ...


-def _target_aggregate(
+def target_aggregate(
    history: Iterable[SectorDailyAggregate], target_trade_date: date
 ) -> SectorDailyAggregate:
+    """Require one target row; missing or duplicate targets are invalid inputs."""
    matches = tuple(row for row in history if row.trade_date == target_trade_date)
    if len(matches) != 1:
        raise ValueError("history must contain exactly one target-date aggregate")
    return matches[0]


-def _quality(row: SectorDailyAggregate) -> MetricQuality:
+def aggregate_quality(row: SectorDailyAggregate) -> MetricQuality:
+    """Retain limited-sample status for incomplete membership or flow coverage."""
    if row.valid_sample_count < 5 or row.membership_coverage < 1 or row.moneyflow_coverage < 1:
        return MetricQuality.AVAILABLE_LIMITED_SAMPLE
    return MetricQuality.AVAILABLE


-def _observation(
+def make_metric_observation(
    row: SectorDailyAggregate,
    *,
    metric_kind: MetricKind,
@@ -57,6 +59,7 @@ def _observation(
    value: Decimal | None,
    quality: MetricQuality | None = None,
 ) -> MetricObservation:
+    """Carry source coverage into a metric and mark unknown raw values unavailable."""
    return MetricObservation(
        trade_date=row.trade_date,
        sector_type=row.sector_type,
@@ -72,7 +75,7 @@ def _observation(
            if value is None
            else quality
            if quality is not None
-            else _quality(row)
+            else aggregate_quality(row)
        ),
        member_count=row.member_count,
        valid_sample_count=row.valid_sample_count,
@@ -95,9 +98,9 @@ class AmountNetStrategy:
    ) -> MetricObservation:
        """Return the target net amount; missing moneyflow remains unavailable."""

-        row = _target_aggregate(history, target_trade_date)
+        row = target_aggregate(history, target_trade_date)
        value = None if row.net_amount_yuan is None else row.net_amount_yuan / Decimal("100000000")
-        return _observation(
+        return make_metric_observation(
            row,
            metric_kind=self.metric_kind,
            metric_version=self.metric_version,
@@ -120,7 +123,7 @@ class RatioTurnoverStrategy:
    ) -> MetricObservation:
        """Return a ratio only when numerator and positive denominator exist."""

-        row = _target_aggregate(history, target_trade_date)
+        row = target_aggregate(history, target_trade_date)
        value = None
        if (
            row.net_amount_yuan is not None
@@ -128,7 +131,7 @@ class RatioTurnoverStrategy:
            and row.turnover_yuan > 0
        ):
            value = row.net_amount_yuan / row.turnover_yuan
-        return _observation(
+        return make_metric_observation(
            row,
            metric_kind=self.metric_kind,
            metric_version=self.metric_version,
@@ -156,7 +159,7 @@ class SwingEqualThreeToTenStrategy:
        """Calculate eight complete trading-day windows ending at the target."""

        rows = tuple(sorted(history, key=lambda row: row.trade_date))
-        target = _target_aggregate(rows, target_trade_date)
+        target = target_aggregate(rows, target_trade_date)
        eligible = tuple(row for row in rows if row.trade_date <= target_trade_date)
        if any(
            (row.sector_type, row.sector_code) != (target.sector_type, target.sector_code)
@@ -190,11 +193,11 @@ class SwingEqualThreeToTenStrategy:
                value = sum(window_ratios, start=Decimal(0)) / Decimal(8)
                quality = (
                    MetricQuality.AVAILABLE_LIMITED_SAMPLE
-                    if any(_quality(row) is not MetricQuality.AVAILABLE for row in latest)
+                    if any(aggregate_quality(row) is not MetricQuality.AVAILABLE for row in latest)
                    else MetricQuality.AVAILABLE
                )

-        return _observation(
+        return make_metric_observation(
            target,
            metric_kind=self.metric_kind,
            metric_version=self.metric_version,
@@ -260,11 +260,15 @@ class MetricObservation:
    valid_sample_count: int
    membership_coverage: Decimal
    moneyflow_coverage: Decimal
+    weighted_score: Decimal | None = None

    def __post_init__(self) -> None:
        """Keep unavailable and finite-value states internally consistent."""

        _validate_finite_decimal(self.value, "value")
+        _validate_finite_decimal(self.weighted_score, "weighted_score")
+        if self.weighted_score is not None and self.value is None:
+            raise ValueError("a weighted score requires an observed raw value")
        if self.value is None and self.quality is not MetricQuality.UNAVAILABLE:
            raise ValueError("a missing metric value must be unavailable")
        if self.value is not None and self.quality is MetricQuality.UNAVAILABLE:
@@ -293,6 +293,12 @@ class SectorRadarRepository(Protocol):

    def save_daily_aggregates(self, records: Iterable[DailyAggregateRecord]) -> WriteCounts: ...

+    def load_publication_aggregates(
+        self, publication_ids: Sequence[str]
+    ) -> Sequence[DailyAggregateRecord]:
+        """Read exact revisions, retaining source details for offline rescoring."""
+        ...
+
    def create_publication(self, publication: RadarPublication) -> WriteCounts: ...

    def finish_publication(self, publication: RadarPublication) -> None: ...
@@ -16,6 +16,7 @@ from .models import (
    RankSide,
    SectorType,
 )
+from .weighted import ranking_value

 PoolKey = tuple[date, SectorType, MetricKind, str]
 SectorMetricKey = tuple[SectorType, str, MetricKind, str]
@@ -40,9 +41,10 @@ def _sector_metric_key(observation: MetricObservation) -> SectorMetricKey:


 def _available_sort_key(observation: MetricObservation) -> tuple[Decimal, str]:
-    if observation.value is None:
+    value = ranking_value(observation)
+    if value is None:
        raise ValueError("unavailable observations cannot use the ranking sort key")
-    return (-observation.value, observation.sector_code)
+    return (-value, observation.sector_code)


 def rank_metric_observations(
@@ -70,7 +72,7 @@ def rank_metric_observations(
            raise ValueError("a ranking pool must not contain duplicate sector codes")

        available = sorted(
-            (observation for observation in pool if observation.value is not None),
+            (observation for observation in pool if ranking_value(observation) is not None),
            key=_available_sort_key,
        )
        pool_size = len(available)
@@ -93,7 +95,7 @@ def rank_metric_observations(
                rank_percentile=None,
            )
            for observation in sorted(
-                (observation for observation in pool if observation.value is None),
+                (observation for observation in pool if ranking_value(observation) is None),
                key=lambda observation: observation.sector_code,
            )
        )
@@ -134,7 +136,7 @@ def select_percentile_side(
            sorted(
                pools[pool_key],
                key=lambda row: (
-                    row.observation.value if row.observation.value is not None else Decimal(0),
+                    ranking_value(row.observation) or Decimal(0),
                    row.observation.sector_code,
                ),
            )
@@ -0,0 +1,246 @@
+"""Versioned flow scores: separate observable ratios from cross-sectional scores."""
+
+from __future__ import annotations
+
+from collections import defaultdict
+from collections.abc import Iterable, Mapping, Sequence
+from dataclasses import dataclass, replace
+from datetime import date
+from decimal import Decimal
+
+from .metrics import (
+    AmountNetStrategy,
+    RatioTurnoverStrategy,
+    SwingEqualThreeToTenStrategy,
+    aggregate_quality,
+    make_metric_observation,
+    target_aggregate,
+)
+from .models import (
+    MetricKind,
+    MetricObservation,
+    MetricQuality,
+    MetricUnit,
+    SectorDailyAggregate,
+    SectorType,
+)
+
+RATIO_WEIGHTED_VERSION = "zhixing_ratio_weighted_v2"
+SWING_WEIGHTED_VERSION = "zhixing_swing_weighted_v2"
+WEIGHTED_VERSIONS = frozenset((RATIO_WEIGHTED_VERSION, SWING_WEIGHTED_VERSION))
+SectorKey = tuple[SectorType, str]
+
+
+def resolve_metric_version(kind: MetricKind, versions: Iterable[str]) -> str | None:
+    """Select a supported version actually present in an immutable publication."""
+    available = set(versions)
+    candidates = {
+        MetricKind.AMOUNT: (AmountNetStrategy.metric_version,),
+        MetricKind.RATIO: (RATIO_WEIGHTED_VERSION, RatioTurnoverStrategy.metric_version),
+        MetricKind.SWING: (SWING_WEIGHTED_VERSION, SwingEqualThreeToTenStrategy.metric_version),
+    }
+    return next((version for version in candidates[kind] if version in available), None)
+
+
+def ranking_value(observation: MetricObservation) -> Decimal | None:
+    """Never substitute a raw ratio for a missing score during v2 warm-up."""
+    return (
+        observation.weighted_score
+        if observation.metric_version in WEIGHTED_VERSIONS
+        else observation.value
+    )
+
+
+def daily_flow_ratio(row: SectorDailyAggregate) -> Decimal | None:
+    """Use yuan consistently, with the reviewed 100-yuan denominator offset."""
+    if row.net_amount_yuan is None or row.turnover_yuan is None or row.turnover_yuan <= 0:
+        return None
+    return row.net_amount_yuan / (row.turnover_yuan + Decimal(100))
+
+
+@dataclass(frozen=True, slots=True)
+class FlowFeatures:
+    """Past-only inputs before any ranking or page filtering is applied."""
+
+    daily_ratio: Decimal | None
+    short_ratio: Decimal | None
+    long_ratio: Decimal | None
+    liquidity_weight: Decimal | None
+    quality: MetricQuality
+
+
+def flow_features(history: Iterable[SectorDailyAggregate], target: date) -> FlowFeatures:
+    """Extract complete 3/5/10-session windows; unknown inputs stay unknown.
+
+    The caller supplies calendar-aligned history, including explicit missing
+    aggregates for calendar holes. Future observations never enter a window.
+    """
+    rows = tuple(
+        sorted((row for row in history if row.trade_date <= target), key=lambda row: row.trade_date)
+    )
+    current = target_aggregate(rows, target)
+    if len({row.trade_date for row in rows}) != len(rows):
+        raise ValueError("history must not contain duplicate trade dates")
+    if any(
+        (row.sector_type, row.sector_code) != (current.sector_type, current.sector_code)
+        for row in rows
+    ):
+        raise ValueError("history must contain exactly one sector identity")
+
+    def mean_ratio(window: int) -> Decimal | None:
+        ratios = tuple(daily_flow_ratio(row) for row in rows[-window:])
+        if len(ratios) < window or any(value is None for value in ratios):
+            return None
+        return sum((value for value in ratios if value is not None), Decimal(0)) / window
+
+    turnovers = tuple(row.turnover_yuan for row in rows[-5:])
+    weight = None
+    if len(turnovers) == 5 and all(value is not None and value > 0 for value in turnovers):
+        average = sum((value for value in turnovers if value is not None), Decimal(0)) / 5
+        weight = (average + 1).log10() / 10
+    quality = (
+        MetricQuality.AVAILABLE_LIMITED_SAMPLE
+        if any(aggregate_quality(row) is not MetricQuality.AVAILABLE for row in rows[-10:])
+        else MetricQuality.AVAILABLE
+    )
+    return FlowFeatures(daily_flow_ratio(current), mean_ratio(3), mean_ratio(10), weight, quality)
+
+
+class RatioWeightedStrategy:
+    """Keep the raw daily ratio; attach its score only after observing the full pool."""
+
+    metric_kind = MetricKind.RATIO
+    metric_version = RATIO_WEIGHTED_VERSION
+    unit = MetricUnit.RATIO
+
+    def evaluate(
+        self, history: Iterable[SectorDailyAggregate], target_trade_date: date
+    ) -> MetricObservation:
+        """Return a daily observation without pretending that a score is a ratio."""
+        rows = tuple(history)
+        current = target_aggregate(rows, target_trade_date)
+        features = flow_features(rows, target_trade_date)
+        return make_metric_observation(
+            current,
+            metric_kind=self.metric_kind,
+            metric_version=self.metric_version,
+            unit=self.unit,
+            value=features.daily_ratio,
+            quality=features.quality,
+        )
+
+
+class SwingWeightedStrategy:
+    """Combine 3- and 10-session simple means, equally weighted."""
+
+    metric_kind = MetricKind.SWING
+    metric_version = SWING_WEIGHTED_VERSION
+    unit = MetricUnit.RATIO
+
+    def evaluate(
+        self, history: Iterable[SectorDailyAggregate], target_trade_date: date
+    ) -> MetricObservation:
+        """Expose the mean ratio independently of the two percentile ranks."""
+        rows = tuple(history)
+        current = target_aggregate(rows, target_trade_date)
+        features = flow_features(rows, target_trade_date)
+        value = (
+            (features.short_ratio + features.long_ratio) / 2
+            if features.short_ratio is not None and features.long_ratio is not None
+            else None
+        )
+        return make_metric_observation(
+            current,
+            metric_kind=self.metric_kind,
+            metric_version=self.metric_version,
+            unit=self.unit,
+            value=value,
+            quality=features.quality,
+        )
+
+
+def calendar_sector_history(
+    current: SectorDailyAggregate,
+    previous: Mapping[date, SectorDailyAggregate],
+    trading_dates: Sequence[date],
+) -> tuple[SectorDailyAggregate, ...]:
+    """Represent missing trading sessions explicitly rather than skipping them."""
+    dates = sorted({day for day in trading_dates if day <= current.trade_date})[-10:]
+    if not dates or dates[-1] != current.trade_date:
+        raise ValueError("the target must belong to the observed trading calendar")
+    return tuple(
+        current
+        if day == current.trade_date
+        else previous.get(day)
+        or replace(
+            current,
+            trade_date=day,
+            member_count=0,
+            valid_sample_count=0,
+            net_amount_yuan=None,
+            turnover_yuan=None,
+            membership_coverage=Decimal(0),
+            moneyflow_coverage=Decimal(0),
+        )
+        for day in dates
+    )
+
+
+def _percentiles(values: Mapping[str, Decimal | None]) -> dict[str, Decimal]:
+    """Compute ascending average-rank percentiles, preserving ties deterministically."""
+    ordered = sorted((value, code) for code, value in values.items() if value is not None)
+    result: dict[str, Decimal] = {}
+    start = 0
+    while start < len(ordered):
+        end = start + 1
+        while end < len(ordered) and ordered[end][0] == ordered[start][0]:
+            end += 1
+        percentile = Decimal(start + 1 + end) / (2 * len(ordered))
+        result.update((code, percentile) for _, code in ordered[start:end])
+        start = end
+    return result
+
+
+def attach_weighted_scores(
+    observations: Sequence[MetricObservation],
+    features: Mapping[SectorKey, FlowFeatures],
+) -> tuple[MetricObservation, ...]:
+    """Score complete date/type/version pools, never a paginated subset.
+
+    Missing history can leave a raw value available without a final rank.
+    The two swing percentiles are calculated separately before combining.
+    """
+    pools: defaultdict[tuple[date, SectorType, str], list[MetricObservation]] = defaultdict(list)
+    for row in observations:
+        if row.metric_version in WEIGHTED_VERSIONS:
+            pools[(row.trade_date, row.sector_type, row.metric_version)].append(row)
+    scores: dict[tuple[date, SectorType, str, str], Decimal | None] = {}
+    for pool in pools.values():
+        feature = {row.sector_code: features[(row.sector_type, row.sector_code)] for row in pool}
+        daily = _percentiles({code: item.daily_ratio for code, item in feature.items()})
+        short = _percentiles({code: item.short_ratio for code, item in feature.items()})
+        long = _percentiles({code: item.long_ratio for code, item in feature.items()})
+        for row in pool:
+            code = row.sector_code
+            weight = feature[code].liquidity_weight
+            percentile = (
+                daily.get(code)
+                if row.metric_kind is MetricKind.RATIO
+                else ((short[code] + long[code]) / 2 if code in short and code in long else None)
+            )
+            scores[(row.trade_date, row.sector_type, code, row.metric_version)] = (
+                Decimal(1000) * percentile * weight
+                if percentile is not None and weight is not None and row.value is not None
+                else None
+            )
+    return tuple(
+        replace(
+            row,
+            weighted_score=scores[
+                (row.trade_date, row.sector_type, row.sector_code, row.metric_version)
+            ],
+        )
+        if row.metric_version in WEIGHTED_VERSIONS
+        else row
+        for row in observations
+    )
@@ -254,6 +254,26 @@ class InMemorySectorRadarRepository:
            ),
        )

+    def load_publication_aggregates(
+        self, publication_ids: Sequence[str]
+    ) -> Sequence[DailyAggregateRecord]:
+        """Load exact immutable aggregate revisions, including detail fields."""
+        wanted = set(publication_ids)
+        return tuple(
+            sorted(
+                (
+                    record
+                    for record in self.daily_aggregates.values()
+                    if record.publication_id in wanted
+                ),
+                key=lambda record: (
+                    record.aggregate.trade_date,
+                    record.aggregate.sector_type,
+                    record.aggregate.sector_code,
+                ),
+            )
+        )
+
    def create_publication(self, publication: RadarPublication) -> WriteCounts:
        """Create one running publication without replacing an existing identity."""

@@ -493,6 +493,34 @@ class PostgresSectorRadarRepository:
            rows,
        )

+    def load_publication_aggregates(
+        self, publication_ids: Sequence[str]
+    ) -> Sequence[DailyAggregateRecord]:
+        """Batch-read exact input revisions without dropping detail provenance."""
+        if not publication_ids:
+            return ()
+        with self._connection() as connection:
+            rows = connection.execute(
+                """
+                SELECT publication_id, trade_date, sector_type, sector_code, sector_name,
+                       member_count, valid_sample_count, net_amount_yuan, turnover_yuan,
+                       membership_coverage, moneyflow_coverage, pct_change, leading_code
+                FROM sector_radar_daily_aggregate
+                WHERE publication_id = ANY(%s)
+                ORDER BY trade_date, sector_type, sector_code
+                """,
+                (list(publication_ids),),
+            ).fetchall()
+        return tuple(
+            DailyAggregateRecord(
+                str(row[0]),
+                self._aggregate_from_row(row[1:11]),
+                None if row[11] is None else Decimal(str(row[11])),
+                None if row[12] is None else str(row[12]),
+            )
+            for row in rows
+        )
+
    def create_publication(self, publication: RadarPublication) -> WriteCounts:
        """Insert a new running publication identity idempotently."""

@@ -694,6 +722,7 @@ class PostgresSectorRadarRepository:
                    "rank_position",
                    "rank_percentile",
                    "rank_changes",
+                    "weighted_score",
                ),
                ("publication_id", "sector_type", "sector_code", "metric_version"),
                self._ranking_rows(ranking_items),
@@ -772,6 +801,7 @@ class PostgresSectorRadarRepository:
                "rank_position",
                "rank_percentile",
                "rank_changes",
+                "weighted_score",
            ),
            ("publication_id", "sector_type", "sector_code", "metric_version"),
            self._ranking_rows(items),
@@ -877,7 +907,8 @@ class PostgresSectorRadarRepository:
                SELECT trade_date, sector_type, sector_code, sector_name, metric_kind,
                       metric_version, implementation_kind, unit, metric_value, quality,
                       member_count, valid_sample_count, membership_coverage,
-                       moneyflow_coverage, rank_position, rank_percentile, rank_changes
+                       moneyflow_coverage, rank_position, rank_percentile, rank_changes,
+                       weighted_score
                FROM sector_radar_ranking
                WHERE publication_id = %s
                ORDER BY sector_type, metric_version, rank_position NULLS LAST, sector_code
@@ -898,7 +929,8 @@ class PostgresSectorRadarRepository:
                SELECT publication_id, trade_date, sector_type, sector_code, sector_name,
                       metric_kind, metric_version, implementation_kind, unit, metric_value,
                       quality, member_count, valid_sample_count, membership_coverage,
-                       moneyflow_coverage, rank_position, rank_percentile, rank_changes
+                       moneyflow_coverage, rank_position, rank_percentile, rank_changes,
+                       weighted_score
                FROM sector_radar_ranking
                WHERE publication_id = ANY(%s)
                ORDER BY publication_id, sector_type, metric_kind, rank_position NULLS LAST,
@@ -938,7 +970,8 @@ class PostgresSectorRadarRepository:
                       ranking.implementation_kind, ranking.unit, ranking.metric_value,
                       ranking.quality, ranking.member_count, ranking.valid_sample_count,
                       ranking.membership_coverage, ranking.moneyflow_coverage,
-                       ranking.rank_position, ranking.rank_percentile, ranking.rank_changes
+                       ranking.rank_position, ranking.rank_percentile, ranking.rank_changes,
+                       ranking.weighted_score
                FROM pools AS pool
                LEFT JOIN sector_radar_ranking AS ranking
                  ON ranking.publication_id = pool.publication_id
@@ -1016,7 +1049,7 @@ class PostgresSectorRadarRepository:
                       ranking.metric_value, ranking.quality, ranking.member_count,
                       ranking.valid_sample_count, ranking.membership_coverage,
                       ranking.moneyflow_coverage, ranking.rank_position,
-                       ranking.rank_percentile, ranking.rank_changes
+                       ranking.rank_percentile, ranking.rank_changes, ranking.weighted_score
                FROM sector_radar_ranking AS ranking
                JOIN selected ON selected.id = ranking.publication_id
                ORDER BY selected.target_trade_date DESC, ranking.sector_type,
@@ -1168,6 +1201,7 @@ class PostgresSectorRadarRepository:
                    ranking.rank_position,
                    ranking.rank_percentile,
                    Jsonb({str(change.days): change.value for change in ranking.rank_changes}),
+                    observation.weighted_score,
                )
            )
        return tuple(rows)
@@ -1258,6 +1292,7 @@ class PostgresSectorRadarRepository:
            valid_sample_count=int(row[11]),
            membership_coverage=Decimal(str(row[12])),
            moneyflow_coverage=Decimal(str(row[13])),
+            weighted_score=None if row[17] is None else Decimal(str(row[17])),
        )
        return RankedMetric(
            observation=observation,
@@ -10,6 +10,7 @@ from datetime import date

 from ....bootstrap.config import get_settings
 from ..application.build import BuildSectorRadar, BuildSectorRadarCommand
+from ..application.recompute import RecomputeSectorRadar
 from ..infrastructure.postgres import PostgresSectorRadarRepository
 from ..infrastructure.tushare import TushareSectorRadarAdapter

@@ -30,6 +31,11 @@ def build_parser() -> argparse.ArgumentParser:
    )
    mode.add_argument("--start-date", type=_parse_date, help="inclusive backfill start date")
    parser.add_argument("--end-date", type=_parse_date, help="inclusive backfill end date")
+    parser.add_argument(
+        "--recompute",
+        action="store_true",
+        help="rescore saved aggregates without fetching external sources",
+    )
    return parser


@@ -39,6 +45,8 @@ def main(argv: Sequence[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    if (args.start_date is None) != (args.end_date is None):
        raise SystemExit("--start-date and --end-date must be provided together")
+    if args.recompute and args.retry_publication_id:
+        raise SystemExit("--recompute cannot be combined with --retry-publication-id")
    command = BuildSectorRadarCommand(
        trade_date=args.trade_date,
        start_date=args.start_date,
@@ -59,22 +67,25 @@ def main(argv: Sequence[str] | None = None) -> int:
            command.end_date or "none",
            bool(command.retry_publication_id),
        )
-        source = TushareSectorRadarAdapter.from_token(
-            settings.tushare_token,
-            max_retries=settings.sector_radar_max_retries,
-            backoff_seconds=settings.sector_radar_retry_backoff_seconds,
-            request_interval_seconds=settings.sector_radar_request_interval_seconds,
-        )
        repository = PostgresSectorRadarRepository(
            settings.database_url,
            advisory_lock_key=settings.sector_radar_advisory_lock_key,
        )
        try:
-            summary = BuildSectorRadar(
-                source,
-                repository,
-                coverage_threshold=settings.sector_radar_coverage_threshold,
-            ).execute(command)
+            if args.recompute:
+                summary = RecomputeSectorRadar(repository).execute(command)
+            else:
+                source = TushareSectorRadarAdapter.from_token(
+                    settings.tushare_token,
+                    max_retries=settings.sector_radar_max_retries,
+                    backoff_seconds=settings.sector_radar_retry_backoff_seconds,
+                    request_interval_seconds=settings.sector_radar_request_interval_seconds,
+                )
+                summary = BuildSectorRadar(
+                    source,
+                    repository,
+                    coverage_threshold=settings.sector_radar_coverage_threshold,
+                ).execute(command)
        finally:
            repository.close()
    except Exception as exc:  # noqa: BLE001 - CLI boundary returns a redacted scheduler result
@@ -98,6 +98,7 @@ class RadarRankingRowResponse(BaseModel):
    implementation_kind: Literal["independent"]
    unit: MetricUnit
    metric_value: Decimal | None
+    weighted_score: Decimal | None = None
    quality: MetricQuality
    member_count: int = Field(ge=0)
    valid_sample_count: int = Field(ge=0)
@@ -107,6 +108,9 @@ class RadarRankingRowResponse(BaseModel):
    rank_percentile: Decimal | None = Field(default=None, gt=0, le=100)
    rank_change_days: int = Field(ge=1, le=5)
    rank_change: int | None
+    rank_change_values: dict[MetricKind, int | None] = Field(
+        default_factory=lambda: dict[MetricKind, int | None]()
+    )
    pct_change: Decimal | None = None
    daily_net_amount_yuan: Decimal | None = None
    daily_ratio: Decimal | None = None
@@ -127,6 +131,7 @@ class RadarHistoryMetricResponse(RadarDetailModel):
    rank_percentile: Decimal | None
    pool_size: int
    metric_value: Decimal | None
+    weighted_score: Decimal | None = None
    missing: bool
    in_top: bool
    in_bottom: bool
@@ -222,6 +227,7 @@ class RadarRankingsResponse(BaseModel):
    view: RadarView
    rank_change_metric: MetricKind
    rank_change_days: int = Field(ge=1, le=5)
+    comparison_trade_date: date | None = None
    side: RankSide
    search: str | None
    publication: RadarPublicationResponse | None
@@ -303,7 +309,7 @@ def get_sector_radar_rankings(
    trade_date: date | None = None,
    sector_type: SectorType = SectorType.CONCEPT,
    view: RadarView = RadarView.AMOUNT,
-    rank_change_metric: MetricKind = MetricKind.AMOUNT,
+    rank_change_metric: MetricKind = MetricKind.SWING,
    rank_change_days: Annotated[int, Query(ge=1, le=5)] = 1,
    side: RankSide = RankSide.ALL,
    search: Annotated[str | None, Query(max_length=100)] = None,
@@ -458,6 +464,7 @@ def _rankings_response(page: RankingPage) -> RadarRankingsResponse:
        view=query.view,
        rank_change_metric=query.rank_change_metric,
        rank_change_days=query.rank_change_days,
+        comparison_trade_date=page.comparison_trade_date,
        side=query.side,
        search=query.search,
        publication=(
@@ -469,7 +476,10 @@ def _rankings_response(page: RankingPage) -> RadarRankingsResponse:
        total=page.total,
        rows=[
            _ranking_response(
-                row, query.rank_change_days, page.extras.get(row.observation.sector_code)
+                row,
+                query.rank_change_days,
+                page.extras.get(row.observation.sector_code),
+                page.rank_change_values.get(row.observation.sector_code),
            )
            for row in page.rows
        ],
@@ -525,7 +535,10 @@ def _definition_response(


 def _ranking_response(
-    row: RankedMetric, rank_change_days: int, extras: RankingExtras | None = None
+    row: RankedMetric,
+    rank_change_days: int,
+    extras: RankingExtras | None = None,
+    changes: dict[MetricKind, int | None] | None = None,
 ) -> RadarRankingRowResponse:
    observation = row.observation
    extras = extras or RankingExtras()
@@ -539,6 +552,7 @@ def _ranking_response(
        implementation_kind=observation.implementation_kind,
        unit=observation.unit,
        metric_value=observation.value,
+        weighted_score=observation.weighted_score,
        quality=observation.quality,
        member_count=observation.member_count,
        valid_sample_count=observation.valid_sample_count,
@@ -548,6 +562,7 @@ def _ranking_response(
        rank_percentile=row.rank_percentile,
        rank_change_days=rank_change_days,
        rank_change=row.rank_change(rank_change_days),
+        rank_change_values={kind: (changes or {}).get(kind) for kind in MetricKind},
        pct_change=extras.pct_change,
        daily_net_amount_yuan=extras.daily_net_amount_yuan,
        daily_ratio=extras.daily_ratio,