diff --git a/README.en.md b/README.en.md
index 15f9596..7127a5d 100644
--- a/README.en.md
+++ b/README.en.md
@@ -55,7 +55,7 @@ It is not an automated money maker, promises no profitability, and does not prov
```mermaid
flowchart LR
Op["Operator / External Agent
Web · ht CLI · TUI · Desktop"] --> API["FastAPI Control Plane
Mission · TokenManager · Probes (/livez, /readyz, /healthz)"]
- API --> DB[("PostgreSQL + pgvector
49 migrations · Mission projections · HET tree · Memory")]
+ API --> DB[("PostgreSQL + pgvector
50 migrations · Mission projections · HET tree · Memory")]
Worker["Worker Loops (13 Loops)
Mission · AVO research · evolution · auto-review · meta-tuning"] --> DB
Worker --> Research["ARC Autonomous Research Loop (AVO + RD-Agent)
HET tree → Co-STEER synthesis → AST gatekeeper → sandbox → baseline"]
Worker --> Evolution["Evolution Engine (default on)
decay scan → budget admission → re-research → race duel → ledger"]
@@ -213,7 +213,7 @@ Single command executes all quality gates: Frontend lint, Vitest, and production
| **Control Plane** | FastAPI 0.122+ / Uvicorn, scoped security tokens, `/livez`, `/readyz`, `/healthz` container probes, cursor SSE |
| **Credential Security**| `TokenRotationService` (SHA-256 single-way hashing, zero plaintext storage, 24h grace period hot rotation, 7-day expiration audit) |
| **Domain & Runtime** | Python 3.12+, Pydantic v2 strict models, modular monolith + ports-and-adapters |
-| **Storage** | SQLAlchemy 2.0 + Alembic (49 migrations), PostgreSQL 14+ with `pgvector` |
+| **Storage** | SQLAlchemy 2.0 + Alembic (50 migrations), PostgreSQL 14+ with `pgvector` |
| **Workers** | asyncio + PostgreSQL SQL lease locking (13 concurrent background loops) |
| **Research Core** | ARC + RD-Agent Hypothesis Evolution Tree (HET) + Co-STEER synthesizer + `ASTGatekeeper` |
| **Target Adapters** | `market_target.v1` and `market-evolution.v1` MCP contracts; BitPro (Crypto) + QuantLab (A-Shares) |
diff --git a/README.md b/README.md
index ca6d486..27248a1 100644
--- a/README.md
+++ b/README.md
@@ -55,7 +55,7 @@ It is not an automated money maker, promises no profitability, and does not prov
```mermaid
flowchart LR
Op["Operator / External Agent
Web · ht CLI · TUI · Desktop"] --> API["FastAPI Control Plane
Mission · TokenManager · Probes (/livez, /readyz, /healthz)"]
- API --> DB[("PostgreSQL + pgvector
49 migrations · Mission projections · HET tree · Memory")]
+ API --> DB[("PostgreSQL + pgvector
50 migrations · Mission projections · HET tree · Memory")]
Worker["Worker Loops (13 Loops)
Mission · AVO research · evolution · auto-review · meta-tuning"] --> DB
Worker --> Research["ARC Autonomous Research Loop (AVO + RD-Agent)
HET tree → Co-STEER synthesis → AST gatekeeper → sandbox → baseline"]
Worker --> Evolution["Evolution Engine (default on)
decay scan → budget admission → re-research → race duel → ledger"]
diff --git a/README.zh-CN.md b/README.zh-CN.md
index 860411a..20d2a2b 100644
--- a/README.zh-CN.md
+++ b/README.zh-CN.md
@@ -55,7 +55,7 @@
```mermaid
flowchart LR
Op["操作员 / 外部 Agent
Web · ht CLI · TUI · Desktop"] --> API["FastAPI 控制面
Mission · TokenManager · Probes (/livez, /readyz, /healthz)"]
- API --> DB[("PostgreSQL + pgvector
49 项迁移 · Mission 投影 · 演进树 · 记忆库")]
+ API --> DB[("PostgreSQL + pgvector
50 项迁移 · Mission 投影 · 演进树 · 记忆库")]
Worker["Worker 循环 (13 个 Loop)
Mission · AVO 研究 · 进化扫描 · 自动评审 · 元学习"] --> DB
Worker --> Research["ARC 自主研究循环 (AVO + RD-Agent)
HET 假设树 → Co-STEER 合成 → AST 门禁 → 沙箱自测 → 基线对比"]
Worker --> Evolution["自主进化引擎（默认开启）
退化扫描 → 预算准入 → 再研究 → 赛马对决 → 效果账本"]
@@ -213,7 +213,7 @@ curl http://localhost:3334/api/portfolio/relay/handovers
| **控制面** | FastAPI 0.122+ / Uvicorn，作用域安全令牌，`/livez`, `/readyz`, `/healthz` 容器探活矩阵，游标可续传 SSE |
| **凭据安全** | `TokenRotationService`（SHA-256 单向散列，零明文存储，24h 宽限期热轮换，7天临期审计） |
| **运行时与领域** | Python 3.12+、Pydantic v2 严格合同，模块化单体 + ports-and-adapters |
-| **存储** | SQLAlchemy 2.0 + Alembic（49 个迁移），PostgreSQL 14+ 带 `pgvector` |
+| **存储** | SQLAlchemy 2.0 + Alembic（50 个迁移），PostgreSQL 14+ 带 `pgvector` |
| **Worker 集群** | asyncio + PostgreSQL SQL lease 租约（13 个并发后台循环） |
| **研究与演进中枢** | ARC + RD-Agent 假设演进树 (HET) + Co-STEER 代码合成器 + `ASTGatekeeper` 语法守卫 |
| **多市场目标适配** | `market_target.v1` 与 `market-evolution.v1` MCP 契约；BitPro (Crypto) + QuantLab (A股) |
diff --git a/backend/alembic/versions/0050_arc_forward_outcomes.py b/backend/alembic/versions/0050_arc_forward_outcomes.py
new file mode 100644
index 0000000..2833bea
--- /dev/null
+++ b/backend/alembic/versions/0050_arc_forward_outcomes.py
@@ -0,0 +1,94 @@
+"""Persist finished BitPro race lane outcomes for research memory."""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0050_arc_forward_outcomes"
+down_revision: str | None = "0049_arc_cognitive_memory"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def _timestamps() -> tuple[sa.Column[object], sa.Column[object]]:
+ return (
+ sa.Column(
+ "created_at",
+ sa.DateTime(timezone=True),
+ nullable=False,
+ server_default=sa.func.now(),
+ ),
+ sa.Column(
+ "updated_at",
+ sa.DateTime(timezone=True),
+ nullable=False,
+ server_default=sa.func.now(),
+ ),
+ )
+
+
+def upgrade() -> None:
+ op.create_table(
+ "arc_forward_outcomes",
+ sa.Column("id", sa.String(64), primary_key=True),
+ sa.Column("target_id", sa.String(32), nullable=False),
+ sa.Column("parent_strategy_id", sa.String(64), nullable=False),
+ sa.Column("challenger_strategy_id", sa.String(64), nullable=False),
+ sa.Column("slot_no", sa.Integer(), nullable=True),
+ sa.Column("generation", sa.Integer(), nullable=False, server_default="1"),
+ sa.Column("verdict", sa.String(32), nullable=False),
+ sa.Column("verdict_source", sa.String(32), nullable=False, server_default="race_judge"),
+ sa.Column("ended_at", sa.DateTime(timezone=True), nullable=True),
+ sa.Column("metrics_json", sa.JSON(), nullable=False, server_default="{}"),
+ sa.Column("evidence_sha256", sa.String(64), nullable=False, server_default=""),
+ sa.Column("lineage_status", sa.String(16), nullable=False, server_default="pending"),
+ sa.Column("mission_id", sa.String(64), nullable=True),
+ sa.Column("candidate_id", sa.String(128), nullable=True),
+ sa.Column("attempt_id", sa.String(128), nullable=True),
+ sa.Column("changes_json", sa.JSON(), nullable=False, server_default="{}"),
+ sa.Column("scope_json", sa.JSON(), nullable=False, server_default="{}"),
+ *_timestamps(),
+ sa.UniqueConstraint(
+ "target_id",
+ "parent_strategy_id",
+ "challenger_strategy_id",
+ name="uq_arc_forward_outcome_pair",
+ ),
+ )
+ op.create_index(
+ "ix_arc_forward_outcomes_parent_strategy_id",
+ "arc_forward_outcomes",
+ ["parent_strategy_id"],
+ )
+ op.create_index(
+ "ix_arc_forward_outcomes_challenger_strategy_id",
+ "arc_forward_outcomes",
+ ["challenger_strategy_id"],
+ )
+ op.create_index("ix_arc_forward_outcomes_verdict", "arc_forward_outcomes", ["verdict"])
+ op.create_index(
+ "ix_arc_forward_outcomes_lineage_status",
+ "arc_forward_outcomes",
+ ["lineage_status"],
+ )
+ op.create_index(
+ "ix_arc_forward_outcomes_mission_id",
+ "arc_forward_outcomes",
+ ["mission_id"],
+ )
+
+
+def downgrade() -> None:
+ op.drop_index("ix_arc_forward_outcomes_mission_id", table_name="arc_forward_outcomes")
+ op.drop_index("ix_arc_forward_outcomes_lineage_status", table_name="arc_forward_outcomes")
+ op.drop_index("ix_arc_forward_outcomes_verdict", table_name="arc_forward_outcomes")
+ op.drop_index(
+ "ix_arc_forward_outcomes_challenger_strategy_id",
+ table_name="arc_forward_outcomes",
+ )
+ op.drop_index(
+ "ix_arc_forward_outcomes_parent_strategy_id",
+ table_name="arc_forward_outcomes",
+ )
+ op.drop_table("arc_forward_outcomes")
diff --git a/backend/src/hypertrade/arc/avo.py b/backend/src/hypertrade/arc/avo.py
index 35794a0..36fd0f6 100644
--- a/backend/src/hypertrade/arc/avo.py
+++ b/backend/src/hypertrade/arc/avo.py
@@ -179,6 +179,18 @@
Use both successful and failed experiments; match instrument, timeframe and window.
For an identical archived experiment, provide repeat_reason explaining new evidence or
why a controlled replication is needed; unchanged repetitions do not establish improvement.
+Evolution memory may start with forward_outcome entries: real BitPro Paper races in this
+strategy's lineage. relation says how each relates to the strategy you are evolving:
+made_current is the race this strategy itself won to become the champion (its changes are
+a direction that worked); variant_of_current is an earlier variant of this strategy;
+made_ancestor and variant_of_ancestor are races of the champions it replaced
+(generations_back counts how far back). Their changes are relative to that race's parent,
+not necessarily to this strategy. verdict: won = adopted after the forward gate;
+lost_futility = eliminated for lagging far behind; lost_exam = failed the one-time final
+exam. Cite them by memory_id (outcome:...). Do not repeat the direction of a lost variant
+of this strategy without new evidence and a repeat_reason; exploring near a winning
+direction is preferred. A single race is a noisy observation, not a causal conclusion,
+and never replaces development evidence.
source_variant_policy.relay.tried_variants lists the changes BitPro already raced, queued or
eliminated for this parent; a proposal identical to one of them is rejected because BitPro never
races the same changes twice, so propose a different direction. relay.slots shows how many of the
diff --git a/backend/src/hypertrade/arc/evidence_view.py b/backend/src/hypertrade/arc/evidence_view.py
index 9767bf3..a9908b5 100644
--- a/backend/src/hypertrade/arc/evidence_view.py
+++ b/backend/src/hypertrade/arc/evidence_view.py
@@ -77,6 +77,8 @@ def build_evidence_view(projection: ARCMissionProjection) -> dict[str, Any]:
research["research"]["evolution"]["order_sample"] = {
k: v for k, v in context.get("orders", {}).items() if k != "fills"
}
+ # Forward races are not development references: they carry no backtest receipt,
+ # so mixing them in would read as development evidence without a backtest id.
research["research"]["evolution"]["memory_references"] = [
{
"mission_id": item.get("mission_id"),
@@ -84,6 +86,23 @@ def build_evidence_view(projection: ARCMissionProjection) -> dict[str, Any]:
"backtest_id": item.get("development", {}).get("backtest_id"),
}
for item in context.get("memory", [])
+ if item.get("type") != "forward_outcome"
+ ]
+ research["research"]["evolution"]["forward_outcome_references"] = [
+ {
+ key: item.get(key)
+ for key in (
+ "memory_id",
+ "verdict",
+ "relation",
+ "generations_back",
+ "parent_strategy_id",
+ "challenger_strategy_id",
+ "ended_at",
+ )
+ }
+ for item in context.get("memory", [])
+ if item.get("type") == "forward_outcome"
]
if projection.goal and projection.goal.paper_review_required:
from hypertrade.arc.paper_review import build_paper_review
diff --git a/backend/src/hypertrade/arc/evolution.py b/backend/src/hypertrade/arc/evolution.py
index d9f4dc4..7b47b8d 100644
--- a/backend/src/hypertrade/arc/evolution.py
+++ b/backend/src/hypertrade/arc/evolution.py
@@ -34,6 +34,7 @@
)
from hypertrade.arc.evolution_models import EvolutionControl, EvolutionCycle
from hypertrade.arc.feedback import collect_windows
+from hypertrade.arc.forward_outcomes import ForwardOutcomeLedger
from hypertrade.arc.relay_seats import relay_seats
from hypertrade.arc.runtime_journal import failure_text, record_runtime_error
from hypertrade.arc.short_horizon import (
@@ -272,6 +273,17 @@ def status(self) -> dict[str, Any]:
"continuations": ContinuationLedger(self.db).view(),
}
+ def forward_outcomes_summary(self) -> dict[str, Any]:
+ """Ledger aggregate for the evolution page; failures stay on the page, not raised.
+
+ Kept out of ``status()`` because the research scan calls that several times per
+ cycle and never reads this telemetry.
+ """
+ try:
+ return ForwardOutcomeLedger(self.db).summary()
+ except Exception as exc: # Read-only telemetry must never take the page down.
+ return {"error": failure_text(exc)[:200]}
+
def configure(self, config: EvolutionConfig, *, revision: int, actor: str) -> dict[str, Any]:
try:
profile = get_market_target(config.target_id).profile
@@ -1100,4 +1112,78 @@ def _memory(self, context: dict[str, Any], now: datetime) -> list[dict[str, Any]
)
context["memory_manifest"] = manifest
context["research_windows"] = windows.model_dump(mode="json")
- return memory
+
+ def forward_outcome_memory(row: dict[str, Any]) -> dict[str, Any]:
+ # One race is evidence about a direction, never a causal conclusion; the
+ # prompt tells the model to treat these as noisy and cite them by memory_id.
+ return {
+ "memory_id": f"outcome:{row['id']}",
+ "type": "forward_outcome",
+ "evidence_status": "forward_paper_outcome",
+ "outcome_id": row["id"],
+ "verdict": row["verdict"],
+ "relation": row["relation"],
+ "generations_back": row["generations_back"],
+ "parent_strategy_id": row["parent_strategy_id"],
+ "challenger_strategy_id": row["challenger_strategy_id"],
+ "mission_id": row["mission_id"],
+ "candidate_id": row["candidate_id"],
+ "changes": row["changes"],
+ "metrics": {
+ key: row["metrics"].get(key)
+ for key in (
+ "excess_return_pct",
+ "parent_net_return_pct",
+ "challenger_net_return_pct",
+ "parent_max_drawdown_pct",
+ "challenger_max_drawdown_pct",
+ "positive_days",
+ "paired_non_tie_days",
+ "observed_hours",
+ "cancellation_reason",
+ )
+ },
+ "ended_at": row["ended_at"],
+ "rule": (
+ "One forward Paper race; changes are relative to that race's parent. "
+ "A noisy observation, not a causal conclusion."
+ ),
+ }
+
+ # Forward results live under the champion-to-champion chain, so they are only
+ # visible after linking and only from the lineage walk. Prepend them: the model
+ # sees the first entries of memory, and the newest races are the most relevant.
+ outcomes: list[dict[str, Any]] = []
+ try:
+ ledger = ForwardOutcomeLedger(self.db)
+ ledger.link_pending(limit=10)
+ rows = ledger.evidence_for_lineage(
+ target_id=context["target_id"],
+ strategy_id=context["source_strategy_id"],
+ ancestors=3,
+ limit=8,
+ )
+ except Exception as exc: # Research continues on development evidence alone.
+ record_runtime_error(
+ "evolution.forward_outcomes",
+ exc,
+ context={
+ "source_strategy_id": str(context["source_strategy_id"]),
+ "target_id": context["target_id"],
+ },
+ db=self.db,
+ )
+ manifest["forward_outcomes"] = {
+ "policy": "forward_outcome_v1",
+ "selected": 0,
+ "error": failure_text(exc)[:200],
+ }
+ else:
+ outcomes = [forward_outcome_memory(row) for row in rows]
+ manifest["forward_outcomes"] = {
+ "policy": "forward_outcome_v1",
+ "selected": len(outcomes),
+ "won": sum(1 for row in rows if row["verdict"] == "won"),
+ "lost": sum(1 for row in rows if row["verdict"] in ("lost_futility", "lost_exam")),
+ }
+ return [*outcomes, *memory]
diff --git a/backend/src/hypertrade/arc/evolution_models.py b/backend/src/hypertrade/arc/evolution_models.py
index d80faa0..7e30ab6 100644
--- a/backend/src/hypertrade/arc/evolution_models.py
+++ b/backend/src/hypertrade/arc/evolution_models.py
@@ -3,7 +3,7 @@
from datetime import datetime
from typing import Any
-from sqlalchemy import JSON, DateTime, Integer, String, Text
+from sqlalchemy import JSON, DateTime, Integer, String, Text, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from hypertrade.db import Base, TimestampMixin
@@ -88,3 +88,34 @@ class ArcModelExchange(Base, TimestampMixin):
tool_calls_json: Mapped[list[Any]] = mapped_column(JSON, default=list)
usage_json: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
truncated_json: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
+
+
+class ArcForwardOutcome(Base, TimestampMixin):
+ """One finished BitPro race lane: what happened to a challenger against its parent."""
+
+ __tablename__ = "arc_forward_outcomes"
+ __table_args__ = (
+ UniqueConstraint(
+ "target_id",
+ "parent_strategy_id",
+ "challenger_strategy_id",
+ name="uq_arc_forward_outcome_pair",
+ ),
+ )
+ id: Mapped[str] = mapped_column(String(64), primary_key=True)
+ target_id: Mapped[str] = mapped_column(String(32))
+ parent_strategy_id: Mapped[str] = mapped_column(String(64), index=True)
+ challenger_strategy_id: Mapped[str] = mapped_column(String(64), index=True)
+ slot_no: Mapped[int | None] = mapped_column(Integer, nullable=True)
+ generation: Mapped[int] = mapped_column(Integer, default=1)
+ verdict: Mapped[str] = mapped_column(String(32), index=True)
+ verdict_source: Mapped[str] = mapped_column(String(32), default="race_judge")
+ ended_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+ metrics_json: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
+ evidence_sha256: Mapped[str] = mapped_column(String(64), default="")
+ lineage_status: Mapped[str] = mapped_column(String(16), default="pending", index=True)
+ mission_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+ candidate_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
+ attempt_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
+ changes_json: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
+ scope_json: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
diff --git a/backend/src/hypertrade/arc/evolution_router.py b/backend/src/hypertrade/arc/evolution_router.py
index 082b870..b0850fe 100644
--- a/backend/src/hypertrade/arc/evolution_router.py
+++ b/backend/src/hypertrade/arc/evolution_router.py
@@ -21,7 +21,14 @@ class EvolutionUpdate(BaseModel):
@router.get("/evolution", dependencies=[Depends(require_scope(ARCScope.READ))])
async def evolution_status(request: Request) -> dict[str, Any]:
- return await run_in_threadpool(EvolutionService(request.app.state.db).status)
+ def _status() -> dict[str, Any]:
+ service = EvolutionService(request.app.state.db)
+ state = service.status()
+ # Page-level telemetry only; the research scan keeps calling status() directly.
+ state["forward_outcomes"] = service.forward_outcomes_summary()
+ return state
+
+ return await run_in_threadpool(_status)
@router.put("/evolution", dependencies=[Depends(require_scope(ARCScope.POLICY))])
diff --git a/backend/src/hypertrade/arc/forward_outcomes.py b/backend/src/hypertrade/arc/forward_outcomes.py
new file mode 100644
index 0000000..614f3b1
--- /dev/null
+++ b/backend/src/hypertrade/arc/forward_outcomes.py
@@ -0,0 +1,524 @@
+"""Forward outcome ledger for finished BitPro race lanes.
+
+Only finished lanes are recorded here. ``won``, ``lost_futility`` and ``lost_exam`` are
+the only verdicts that count as evidence of a win or a loss; every other verdict is
+neutral bookkeeping (a timeout, a seat handed to a sibling, an operator cancel, or a
+state BitPro did not explain) and must never be read as negative evidence. One race is a
+noisy observation about one challenger, not a causal conclusion about its strategy
+family.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import math
+from datetime import datetime
+from typing import Any
+
+from sqlalchemy import case, func, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from hypertrade.arc.contracts import ARCCandidateAttemptV1
+from hypertrade.arc.evolution_models import ArcForwardOutcome
+from hypertrade.arc.store import load_projection
+from hypertrade.db import ArcMission, ArcMissionEvent, Database
+
+logger = logging.getLogger("hypertrade.arc.forward_outcomes")
+
+EVIDENCE_VERDICTS = ("won", "lost_futility", "lost_exam")
+NEUTRAL_VERDICTS = (
+ "expired_timeout",
+ "superseded",
+ "cancelled_operator",
+ "cancelled_unknown",
+ "ended_unknown",
+)
+# Only these two mean "the lane ended, but we do not know why"; a concrete verdict may
+# replace them. Every other verdict is final so a later noisy observation cannot rewrite it.
+UPGRADABLE_VERDICTS = ("cancelled_unknown", "ended_unknown")
+
+# The only events that can bind a BitPro strategy id to an ARC attempt, so scoping the
+# lineage query to them keeps the scan small even on a busy research mission.
+_BITPRO_ID_EVENTS = ("candidate_proposed", "bitpro_self_tested", "candidate_validated")
+
+_NUMERIC_METRIC_KEYS = (
+ "excess_return_pct",
+ "parent_net_return_pct",
+ "challenger_net_return_pct",
+ "parent_max_drawdown_pct",
+ "challenger_max_drawdown_pct",
+ "parent_trades",
+ "challenger_trades",
+ "observed_hours",
+ "sign_test_p",
+ "gate_alpha",
+ "positive_days",
+ "paired_non_tie_days",
+ "slot_no",
+)
+_METRIC_KEYS = (*_NUMERIC_METRIC_KEYS, "cancellation_reason", "cancellation_metrics")
+_CANCELLATION_KINDS = {
+ "futility": "lost_futility",
+ "exam": "lost_exam",
+ "timeout": "expired_timeout",
+ "superseded": "superseded",
+ "operator": "cancelled_operator",
+}
+
+
+def outcome_id(target_id: str, parent_strategy_id: object, challenger_strategy_id: object) -> str:
+ """Stable id for one (target, parent, challenger) lane; str() keeps 41 and "41" equal."""
+ body = json.dumps(
+ [target_id, str(parent_strategy_id), str(challenger_strategy_id)],
+ separators=(",", ":"),
+ )
+ return "fo_" + hashlib.sha256(body.encode()).hexdigest()[:32]
+
+
+def verdict_for(state: str | None, cancellation_kind: str | None) -> str | None:
+ """Map a race judge state to a ledger verdict; None means the lane has not finished."""
+ if state == "completed":
+ return "won"
+ if state == "cancelled":
+ # BitPro's cancellation kind is the only source of truth for who ended the lane;
+ # an unrecognized kind must stay unknown rather than being guessed as negative.
+ kind = cancellation_kind if isinstance(cancellation_kind, str) else ""
+ return _CANCELLATION_KINDS.get(kind, "cancelled_unknown")
+ if state == "ended":
+ return "ended_unknown"
+ return None
+
+
+def _finite_number(value: Any) -> int | float | None:
+ """Evidence keeps its original value; bools and non-finite floats are not results."""
+ if value is None or isinstance(value, bool):
+ return None
+ if not isinstance(value, (int, float)):
+ return None
+ try:
+ return value if math.isfinite(value) else None
+ except OverflowError:
+ return None
+
+
+def metrics_from_record(record: dict[str, Any]) -> dict[str, Any]:
+ """Project a ``RacePairRecord.to_dict()`` into the bounded metrics the ledger keeps."""
+ metrics: dict[str, Any] = {
+ key: _finite_number(record.get(key)) for key in _NUMERIC_METRIC_KEYS
+ }
+ # The judge initializes the day counters to 0 when it has no proof yet, so a zero
+ # non-tie count means "unknown", not "observed zero wins".
+ if not metrics["paired_non_tie_days"]:
+ metrics["positive_days"] = None
+ metrics["paired_non_tie_days"] = None
+ reason = record.get("cancellation_reason")
+ metrics["cancellation_reason"] = reason if isinstance(reason, str) else None
+ raw_metrics = record.get("cancellation_metrics")
+ metrics["cancellation_metrics"] = (
+ dict(raw_metrics) if isinstance(raw_metrics, dict) else None
+ )
+ return {key: metrics[key] for key in _METRIC_KEYS}
+
+
+def _evidence_sha256(verdict: str, metrics: dict[str, Any]) -> str:
+ body = json.dumps({"verdict": verdict, "metrics": metrics}, sort_keys=True, default=str)
+ return hashlib.sha256(body.encode()).hexdigest()
+
+
+def _iso(value: datetime | None) -> str | None:
+ return value.isoformat() if value is not None else None
+
+
+def _apply_lineage(
+ row: ArcForwardOutcome,
+ match: tuple[ArcMission, ARCCandidateAttemptV1] | None,
+) -> None:
+ """Write the lineage verdict once; an unmatched challenger is never inferred."""
+ if match is None:
+ row.lineage_status = "no_lineage"
+ return
+ mission, attempt = match
+ spec = attempt.strategy_spec
+ row.lineage_status = "linked"
+ row.mission_id = mission.mission_id
+ row.candidate_id = attempt.candidate_id
+ row.attempt_id = attempt.attempt_id
+ row.changes_json = {
+ "parameter_changes": spec.get("parameter_changes") or {},
+ "structural_changes": spec.get("structural_changes") or {},
+ }
+ row.scope_json = {
+ "symbol": spec.get("symbol"),
+ "symbols": spec.get("symbols"),
+ "timeframe": spec.get("timeframe"),
+ "parent_strategy_id": spec.get("parent_strategy_id"),
+ }
+
+
+def _event_bitpro_id(event_type: str, payload_id: Any, attempt_id: Any) -> str | None:
+ """The BitPro id an event bound, taken from the JSON paths the query projected."""
+ value = attempt_id if event_type == "candidate_proposed" else payload_id
+ return None if value is None else str(value)
+
+
+class ForwardOutcomeLedger:
+ """Write-once outcome rows plus best-effort lineage back to the research attempt."""
+
+ def __init__(
+ self,
+ db: Database,
+ *,
+ mission_scan_limit: int = 500,
+ event_scan_limit: int = 50000,
+ ) -> None:
+ self.db = db
+ self.mission_scan_limit = mission_scan_limit
+ self.event_scan_limit = event_scan_limit
+
+ def record(
+ self,
+ *,
+ target_id: str,
+ parent_strategy_id: object,
+ challenger_strategy_id: object,
+ verdict: str,
+ metrics: dict[str, Any],
+ ended_at: datetime | None = None,
+ slot_no: int | None = None,
+ generation: int = 1,
+ source: str = "race_judge",
+ link: bool = True,
+ ) -> dict[str, Any]:
+ """Record one finished lane; ``link=False`` leaves it pending for a batched pass."""
+ if verdict not in EVIDENCE_VERDICTS and verdict not in NEUTRAL_VERDICTS:
+ raise ValueError(f"unknown forward outcome verdict: {verdict}")
+ oid = outcome_id(target_id, parent_strategy_id, challenger_strategy_id)
+ digest = _evidence_sha256(verdict, metrics)
+
+ with self.db.session() as session:
+ row = session.get(ArcForwardOutcome, oid)
+ if row is None:
+ row = ArcForwardOutcome(
+ id=oid,
+ target_id=target_id,
+ parent_strategy_id=str(parent_strategy_id),
+ challenger_strategy_id=str(challenger_strategy_id),
+ slot_no=slot_no,
+ generation=generation,
+ verdict=verdict,
+ verdict_source=source,
+ ended_at=ended_at,
+ metrics_json=metrics,
+ evidence_sha256=digest,
+ lineage_status="pending",
+ changes_json={},
+ scope_json={},
+ )
+ session.add(row)
+ try:
+ session.flush()
+ except IntegrityError:
+ # Another writer inserted this lane first: the unique pair wins, and
+ # this call reconciles against the row that actually landed.
+ session.rollback()
+ row = session.get(ArcForwardOutcome, oid)
+ if row is None:
+ raise
+ self._reconcile(
+ row,
+ verdict=verdict,
+ metrics=metrics,
+ digest=digest,
+ ended_at=ended_at,
+ source=source,
+ )
+
+ if link:
+ linked = None
+ try:
+ linked = self.link(oid)
+ except Exception: # Lineage is research plumbing; outcome capture must not fail.
+ logger.warning("Forward outcome lineage link failed for %s", oid, exc_info=True)
+ if linked is not None:
+ return linked
+ with self.db.session() as session:
+ row = session.get(ArcForwardOutcome, oid)
+ if row is None:
+ raise RuntimeError(f"forward outcome disappeared after record: {oid}")
+ return self.view(row)
+
+ @staticmethod
+ def _reconcile(
+ row: ArcForwardOutcome,
+ *,
+ verdict: str,
+ metrics: dict[str, Any],
+ digest: str,
+ ended_at: datetime | None,
+ source: str,
+ ) -> None:
+ """Keep the first concrete verdict; only an unknown ending may be refined later."""
+ if row.verdict == verdict:
+ return
+ if row.verdict in UPGRADABLE_VERDICTS and verdict not in UPGRADABLE_VERDICTS:
+ row.verdict = verdict
+ row.metrics_json = metrics
+ row.evidence_sha256 = digest
+ row.ended_at = ended_at
+ row.verdict_source = source
+
+ def _locate_by_events(
+ self,
+ session: Session,
+ wanted: set[str],
+ ) -> dict[str, str]:
+ """Newest mission id per wanted challenger, read through the event index only.
+
+ Only the two scalar JSON paths are selected: ``candidate_proposed`` payloads carry
+ the whole strategy code, and fetching those rows would put megabytes through the
+ driver on every lineage pass. ``yield_per`` is an execution option so PostgreSQL
+ streams the scan instead of buffering every matching event in the driver.
+ """
+ located: dict[str, str] = {}
+ if not wanted:
+ return located
+ payload_id = ArcMissionEvent.event_json[("payload", "bitpro_strategy_id")].as_string()
+ attempt_id = ArcMissionEvent.event_json[
+ ("payload", "attempt", "bitpro_strategy_id")
+ ].as_string()
+ statement = (
+ select(ArcMissionEvent.mission_id, ArcMissionEvent.event_type, payload_id, attempt_id)
+ .where(ArcMissionEvent.event_type.in_(_BITPRO_ID_EVENTS))
+ .order_by(ArcMissionEvent.id.desc())
+ .limit(self.event_scan_limit)
+ .execution_options(yield_per=500)
+ )
+ with session.execute(statement) as result:
+ for mission_id, event_type, payload_value, attempt_value in result:
+ challenger = _event_bitpro_id(event_type, payload_value, attempt_value)
+ if challenger is None or challenger not in wanted or challenger in located:
+ continue
+ located[challenger] = str(mission_id)
+ if len(located) == len(wanted):
+ break
+ return located
+
+ def _lineage_index(
+ self,
+ session: Session,
+ wanted: set[str],
+ ) -> dict[str, tuple[ArcMission, ARCCandidateAttemptV1]]:
+ """Newest mission per wanted challenger, located by event before scanning missions.
+
+ Research now runs many rounds a day, so the mission that launched a variant can
+ fall out of the recent-mission window before its race ends. The event journal
+ still points at it, so it is queried first; the mission scan remains only for
+ early missions whose events live inline in ``projection_json``.
+ """
+ found: dict[str, tuple[ArcMission, ARCCandidateAttemptV1]] = {}
+ if not wanted:
+ return found
+
+ # Step 1: newest event per challenger.
+ located = self._locate_by_events(session, wanted)
+
+ # Step 2: load each hit mission once; the last matching attempt is the newest one.
+ wanted_by_mission: dict[str, list[str]] = {}
+ for challenger, mission_id in located.items():
+ wanted_by_mission.setdefault(mission_id, []).append(challenger)
+ indexed: set[str] = set()
+ for mission_id, challengers in wanted_by_mission.items():
+ mission = session.get(ArcMission, mission_id)
+ if mission is None:
+ continue
+ indexed.add(mission_id)
+ projection = load_projection(session, mission)
+ attempts: dict[str, ARCCandidateAttemptV1] = {}
+ for attempt in projection.attempts:
+ attempts[str(attempt.bitpro_strategy_id)] = attempt
+ for challenger in challengers:
+ matched = attempts.get(challenger)
+ if matched is not None:
+ found[challenger] = (mission, matched)
+ if len(found) == len(wanted):
+ return found
+
+ # Step 3: fall back to the recent-mission scan for ids only present inline.
+ missing = wanted - found.keys()
+ missions = session.scalars(
+ select(ArcMission)
+ .order_by(ArcMission.updated_at.desc())
+ .limit(self.mission_scan_limit)
+ ).yield_per(20)
+ for mission in missions:
+ if mission.mission_id in indexed:
+ continue
+ projection = load_projection(session, mission)
+ for attempt in projection.attempts:
+ challenger = str(attempt.bitpro_strategy_id)
+ if challenger in missing and challenger not in found:
+ found[challenger] = (mission, attempt)
+ if len(found) == len(wanted):
+ return found
+ return found
+
+ def link(self, outcome_id: str) -> dict[str, Any] | None:
+ """Attach one pending outcome to the ARC attempt that launched its challenger."""
+ with self.db.session() as session:
+ row = session.get(ArcForwardOutcome, outcome_id)
+ if row is None:
+ return None
+ if row.lineage_status != "pending":
+ return self.view(row)
+
+ challenger = str(row.challenger_strategy_id)
+ match = self._lineage_index(session, {challenger}).get(challenger)
+ _apply_lineage(row, match)
+ session.flush()
+ return self.view(row)
+
+ def link_pending(self, limit: int = 20) -> int:
+ """Match up to ``limit`` pending rows with one mission scan for the whole batch."""
+ with self.db.session() as session:
+ rows = list(
+ session.scalars(
+ select(ArcForwardOutcome)
+ .where(ArcForwardOutcome.lineage_status == "pending")
+ .order_by(ArcForwardOutcome.created_at.asc())
+ .limit(max(0, limit))
+ )
+ )
+ if not rows:
+ return 0
+ wanted = {str(row.challenger_strategy_id) for row in rows}
+ index = self._lineage_index(session, wanted)
+ for row in rows:
+ _apply_lineage(row, index.get(str(row.challenger_strategy_id)))
+ session.flush()
+ return len(rows)
+
+ def evidence_for_lineage(
+ self,
+ *,
+ target_id: str,
+ strategy_id: object,
+ ancestors: int = 3,
+ limit: int = 8,
+ ) -> list[dict[str, Any]]:
+ """Concrete, lineage-linked races of this strategy and the champions it replaced.
+
+ A challenger that wins becomes the new champion under a new strategy id, while its
+ winning race (and its predecessors' races) stay recorded under the old id. Walking
+ the ``won`` rows backwards is the only way research on the current champion can see
+ the races that made it and its ancestors. Neutral records never qualify, and
+ ``ancestors=0`` reads only the races recorded under this id.
+ """
+ ordering = (
+ case((ArcForwardOutcome.ended_at.is_(None), 1), else_=0),
+ ArcForwardOutcome.ended_at.desc(),
+ ArcForwardOutcome.created_at.desc(),
+ )
+ chain = [str(strategy_id)]
+ with self.db.session() as session:
+ for _ in range(max(0, ancestors)):
+ won = session.scalars(
+ select(ArcForwardOutcome)
+ .where(
+ ArcForwardOutcome.target_id == target_id,
+ ArcForwardOutcome.challenger_strategy_id == chain[-1],
+ ArcForwardOutcome.verdict == "won",
+ )
+ .order_by(*ordering)
+ .limit(1)
+ ).first()
+ if won is None:
+ break
+ parent = str(won.parent_strategy_id)
+ if parent in chain:
+ break # A cycle is operator-visible corruption, not more history.
+ chain.append(parent)
+ statement = (
+ select(ArcForwardOutcome)
+ .where(
+ ArcForwardOutcome.target_id == target_id,
+ ArcForwardOutcome.parent_strategy_id.in_(chain),
+ ArcForwardOutcome.verdict.in_(EVIDENCE_VERDICTS),
+ ArcForwardOutcome.lineage_status == "linked",
+ )
+ .order_by(*ordering)
+ .limit(max(0, limit))
+ )
+ rows = list(session.scalars(statement).all())
+ results = []
+ for row in rows:
+ item = self.view(row)
+ generations_back = chain.index(row.parent_strategy_id)
+ item["generations_back"] = generations_back
+ item["relation"] = self._relation(row, chain, generations_back)
+ results.append(item)
+ return results
+
+ @staticmethod
+ def _relation(row: ArcForwardOutcome, chain: list[str], generations_back: int) -> str:
+ """Why this row matters to the strategy under research, never a causal claim."""
+ if row.verdict == "won":
+ if row.challenger_strategy_id == chain[0]:
+ return "made_current"
+ if row.challenger_strategy_id in chain:
+ return "made_ancestor"
+ if generations_back == 0:
+ return "variant_of_current"
+ return "variant_of_ancestor"
+
+ def summary(self) -> dict[str, Any]:
+ """Aggregate in the database: this is polled by pages that do not need the metrics."""
+ with self.db.session() as session:
+ verdict_counts = session.execute(
+ select(ArcForwardOutcome.verdict, func.count()).group_by(
+ ArcForwardOutcome.verdict
+ )
+ ).all()
+ lineage_counts = session.execute(
+ select(ArcForwardOutcome.lineage_status, func.count()).group_by(
+ ArcForwardOutcome.lineage_status
+ )
+ ).all()
+ last_recorded_at = session.scalar(select(func.max(ArcForwardOutcome.created_at)))
+ by_verdict = {str(verdict): int(count) for verdict, count in verdict_counts}
+ by_lineage = {str(status): int(count) for status, count in lineage_counts}
+ return {
+ "total": sum(by_verdict.values()),
+ "by_verdict": by_verdict,
+ "by_lineage": by_lineage,
+ "evidence": sum(
+ count for verdict, count in by_verdict.items() if verdict in EVIDENCE_VERDICTS
+ ),
+ "last_recorded_at": _iso(last_recorded_at),
+ }
+
+ @staticmethod
+ def view(row: ArcForwardOutcome) -> dict[str, Any]:
+ return {
+ "id": row.id,
+ "target_id": row.target_id,
+ "parent_strategy_id": row.parent_strategy_id,
+ "challenger_strategy_id": row.challenger_strategy_id,
+ "slot_no": row.slot_no,
+ "generation": row.generation,
+ "verdict": row.verdict,
+ "verdict_source": row.verdict_source,
+ "ended_at": _iso(row.ended_at),
+ "metrics": dict(row.metrics_json or {}),
+ "evidence_sha256": row.evidence_sha256,
+ "lineage_status": row.lineage_status,
+ "mission_id": row.mission_id,
+ "candidate_id": row.candidate_id,
+ "attempt_id": row.attempt_id,
+ "changes": dict(row.changes_json or {}),
+ "scope": dict(row.scope_json or {}),
+ "created_at": _iso(row.created_at),
+ "updated_at": _iso(row.updated_at),
+ }
diff --git a/backend/src/hypertrade/paper/race_judge.py b/backend/src/hypertrade/paper/race_judge.py
index 9e11f00..929d4f1 100644
--- a/backend/src/hypertrade/paper/race_judge.py
+++ b/backend/src/hypertrade/paper/race_judge.py
@@ -25,6 +25,11 @@
import httpx
+from hypertrade.arc.forward_outcomes import (
+ ForwardOutcomeLedger,
+ metrics_from_record,
+ verdict_for,
+)
from hypertrade.bitpro.mcp import BitProToolAdapter
from hypertrade.config import Settings, get_settings
from hypertrade.db import Database
@@ -80,6 +85,17 @@ def _opt_text(value: Any) -> str | None:
return value.strip()[:400] or None
+def _opt_datetime(value: Any) -> datetime | None:
+ """BitPro timestamps are ISO strings; a naive one is UTC, anything else is unknown."""
+ if not isinstance(value, str):
+ return None
+ try:
+ parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+ except ValueError:
+ return None
+ return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC)
+
+
def _pct(value: float | None, signed: bool = True) -> str:
if value is None:
return "未知"
@@ -179,6 +195,8 @@ class RacePairRecord:
gate_alpha: float | None = None
# True once the target listed this pair as a plan; if it later disappears the lane "ended".
plan_seen: bool = False
+ # True once the finished race was written to the forward outcome ledger.
+ outcome_recorded: bool = False
# Who ended a cancelled lane and what BitPro attached; all None while racing or unknown.
cancellation_kind: str | None = None
cancellation_actor: str | None = None
@@ -244,6 +262,7 @@ def _parse_id(val: Any) -> str | int:
slot_no=_slot_no(data.get("slot_no")),
gate_alpha=_opt_float(data.get("gate_alpha")),
plan_seen=bool(data.get("plan_seen", False)),
+ outcome_recorded=bool(data.get("outcome_recorded", False)),
cancellation_kind=_cancellation_kind(data.get("cancellation_kind")),
cancellation_actor=_opt_text(data.get("cancellation_actor")),
cancellation_reason=_opt_text(data.get("cancellation_reason")),
@@ -528,6 +547,8 @@ def __init__(
float(getattr(self.settings, "race_judge_observing_interval_seconds", 900)),
)
self._records: dict[str, RacePairRecord] = {}
+ # Finished lanes saved before the ledger existed are backfilled once per process.
+ self._outcomes_backfilled = False
self._load_state()
def _pair_key(
@@ -648,6 +669,8 @@ def evaluate_pair(
lookup_target = target_id or "bitpro"
key = self._pair_key(lookup_target, parent_id, challenger_id)
record = self._records.get(key)
+ # A finished pair that becomes live again is a new race on the same challenger.
+ previous_state = record.state if record is not None else None
resolved_target = (
target_id
or (record.target_id if record is not None else None)
@@ -698,12 +721,30 @@ def evaluate_pair(
plan = status_data.get("plan")
if plan is None and record.plan_seen and record.state in _LIVE_PLAN_STATES:
+ archived = status_data.get("cancellation")
+ if isinstance(archived, dict):
+ # BitPro eliminated or cancelled this lane and its seat was refilled before the
+ # next poll, so the cancelled plan row is already gone. The archive is what the
+ # race actually ended with: mirror it and announce the elimination the judge
+ # would have seen one step earlier. Never read this response's proof: with no
+ # plan BitPro evaluates the pair as a fresh candidate, not as this race, so the
+ # last mirrored numbers stay.
+ record.state = "cancelled"
+ record.action_taken = None
+ self._mirror_cancellation(record, archived)
+ self._announce(record, auto_enabled=False)
+ if not record.outcome_recorded:
+ self._record_outcome(record, ended_at=archived.get("cancelled_at"))
+ return self._finish_evaluation(record)
+
# The target no longer keeps this lane (its seat went to a new challenger, or the
# plan was cleaned up). Fresh evidence for a pair with no plan would be a
# candidate evaluation, not this race, so keep the last mirrored evidence and
# stop polling. Whatever ended the lane was announced by BitPro itself.
record.state = "ended"
record.reason = "BitPro 已不再保留该赛道（位置已被新的挑战者接管，或计划已被清理）"
+ if not record.outcome_recorded:
+ self._record_outcome(record, ended_at=None)
return self._finish_evaluation(record)
proof = status_data.get("proof") or (plan or {}).get("proof") or {}
@@ -753,17 +794,46 @@ def evaluate_pair(
auto_enabled = False
record.action_taken = None
+ if previous_state in _TERMINAL_STATES and record.state in _LIVE_PLAN_STATES:
+ # The ledger keeps the first concrete verdict for a (parent, challenger) pair, so a
+ # second race by the same challenger cannot overwrite the first: a known limitation.
+ record.outcome_recorded = False
+
if record.state in _LIVE_PLAN_STATES:
self._reset_finished_race(record)
elif record.state == "cancelled":
raw = status_data.get("cancellation") or (plan or {}).get("cancellation")
- fields = raw if isinstance(raw, dict) else {}
- record.cancellation_kind = _cancellation_kind(raw)
- record.cancellation_actor = _opt_text(fields.get("actor"))
- record.cancellation_reason = _opt_text(fields.get("reason"))
- record.cancellation_metrics = _cancellation_metrics(raw)
- record.cancellation_cooldown_hours = _cancellation_cooldown_hours(raw)
+ self._mirror_cancellation(record, raw)
+
+ self._announce(record, auto_enabled=auto_enabled)
+
+ if not record.outcome_recorded:
+ if record.state == "cancelled":
+ cancellation = status_data.get("cancellation")
+ if not isinstance(cancellation, dict):
+ cancellation = (plan or {}).get("cancellation")
+ ended_at = (
+ cancellation.get("cancelled_at") if isinstance(cancellation, dict) else None
+ )
+ self._record_outcome(record, ended_at=ended_at)
+ elif record.state == "completed":
+ self._record_outcome(
+ record, ended_at=plan.get("updated_at") if isinstance(plan, dict) else None
+ )
+
+ return self._finish_evaluation(record)
+ def _mirror_cancellation(self, record: RacePairRecord, raw: Any) -> None:
+ """Mirror who ended the lane; missing or malformed fields stay unknown."""
+ fields = raw if isinstance(raw, dict) else {}
+ record.cancellation_kind = _cancellation_kind(raw)
+ record.cancellation_actor = _opt_text(fields.get("actor"))
+ record.cancellation_reason = _opt_text(fields.get("reason"))
+ record.cancellation_metrics = _cancellation_metrics(raw)
+ record.cancellation_cooldown_hours = _cancellation_cooldown_hours(raw)
+
+ def _announce(self, record: RacePairRecord, *, auto_enabled: bool) -> None:
+ """Send the one lifecycle card this state earns; never adopt or mutate the target."""
if record.state == "observing" and record.eligible and record.proof_sha256:
milestone = "ELIGIBLE" if auto_enabled else "REQUIRES_ADMIN"
record.requires_admin_authorization = not auto_enabled
@@ -796,7 +866,74 @@ def evaluate_pair(
if ok:
record.feishu_notified_states.append(milestone)
- return self._finish_evaluation(record)
+ def _record_outcome(
+ self,
+ record: RacePairRecord,
+ *,
+ ended_at: str | None,
+ source: str = "race_judge",
+ link: bool = True,
+ ) -> bool:
+ """Best-effort ledger write.
+
+ The judge only reads target state and notifies, so a missing or failing ledger must
+ never break the card, the state file or the poll: on failure ``outcome_recorded``
+ stays False and the next evaluation retries.
+ """
+ if self.db is None:
+ return False
+ verdict = verdict_for(record.state, record.cancellation_kind)
+ if verdict is None:
+ return False
+ try:
+ ForwardOutcomeLedger(self.db).record(
+ target_id=record.target_id,
+ parent_strategy_id=record.parent_strategy_id,
+ challenger_strategy_id=record.challenger_strategy_id,
+ verdict=verdict,
+ metrics=metrics_from_record(record.to_dict()),
+ ended_at=_opt_datetime(ended_at),
+ slot_no=record.slot_no,
+ generation=record.generation,
+ source=source,
+ link=link,
+ )
+ except Exception:
+ logger.warning(
+ "Failed to record forward outcome for (%s, %s): %s",
+ record.parent_strategy_id,
+ record.challenger_strategy_id,
+ verdict,
+ exc_info=True,
+ )
+ return False
+ record.outcome_recorded = True
+ return True
+
+ def backfill_outcomes(self) -> int:
+ """Record finished lanes saved before the ledger existed; called once per process."""
+ if self.db is None:
+ return 0
+ recorded = 0
+ try:
+ for record in self._records.values():
+ if record.outcome_recorded or record.state not in (
+ "completed",
+ "cancelled",
+ "ended",
+ ):
+ continue
+ if self._record_outcome(record, ended_at=None, source="backfill", link=False):
+ recorded += 1
+ if recorded:
+ try:
+ ForwardOutcomeLedger(self.db).link_pending(limit=recorded)
+ except Exception:
+ logger.warning("Forward outcome backfill lineage link failed", exc_info=True)
+ self._persist_state()
+ except Exception:
+ logger.warning("Forward outcome backfill failed", exc_info=True)
+ return recorded
def _reset_finished_race(self, record: RacePairRecord) -> None:
"""A re-admitted challenger races again: the earlier cancellation no longer holds."""
@@ -836,6 +973,12 @@ def scan_and_judge_all(self, *, force: bool = False) -> list[RacePairRecord]:
Lanes that are only accumulating evidence are skipped until their next poll time;
`force` (the operator's "run now") looks at every pair.
"""
+ if not self._outcomes_backfilled:
+ self._outcomes_backfilled = True
+ try:
+ self.backfill_outcomes()
+ except Exception:
+ logger.warning("Forward outcome backfill failed", exc_info=True)
plans = self._list_plans()
# The plan listing already says which lanes BitPro cancelled: look at those now
# instead of waiting out their slow poll, so the card arrives promptly.
diff --git a/docs/agent_architecture_and_interview_handbook.md b/docs/agent_architecture_and_interview_handbook.md
index 978c1a4..5b530c1 100644
--- a/docs/agent_architecture_and_interview_handbook.md
+++ b/docs/agent_architecture_and_interview_handbook.md
@@ -80,7 +80,7 @@ HyperTrade 整体系统划分为清晰的七层架构，实现了控制面、计
|
+---------------------------------------------------------------------------------------------------+
| 7. 持久化与审计基座 (Persistence & Telemetry Ledger) |
-| - PostgreSQL 14+ / pgvector (49 项 Alembic 迁移) · Langfuse 分布式追踪 · Agent 飞行记录仪 |
+| - PostgreSQL 14+ / pgvector (50 项 Alembic 迁移) · Langfuse 分布式追踪 · Agent 飞行记录仪 |
+---------------------------------------------------------------------------------------------------+
```
diff --git a/docs/architecture/33-system-architecture.md b/docs/architecture/33-system-architecture.md
index 46c6738..905a12a 100644
--- a/docs/architecture/33-system-architecture.md
+++ b/docs/architecture/33-system-architecture.md
@@ -87,7 +87,7 @@ flowchart TB
end
subgraph Persistence["持久化与审计账本"]
- PG[("PostgreSQL 14+ / pgvector
49 项 Alembic 迁移 · 投影 · 演进树 · 记忆库")]
+ PG[("PostgreSQL 14+ / pgvector
50 项 Alembic 迁移 · 投影 · 演进树 · 记忆库")]
end
Web --> API
@@ -113,9 +113,9 @@ flowchart TB
| **Control Plane** | `hypertrade/main.py`, `hypertrade/security/token_manager.py` | 暴露 REST/SSE 控制端点，分级探活（`/livez`, `/readyz`, `/healthz`），零信任令牌签发/轮换/鉴权 |
| **Domain Core** | `hypertrade/runtime/domain/`, `hypertrade/arc/contracts.py` | 严格定义 Mission、Plan、Step、Capability、HET 假设、NettingPlan 等不可变 Pydantic 模型 |
| **Research Engine** | `hypertrade/arc/`, `hypertrade/research/` | RD-Agent 假设演进、Co-STEER 代码生成、`ASTGatekeeper` 语法守卫、`QuantLabStrategyTranspiler` 转译 |
-| **Paper & Evolution** | `hypertrade/arc/evolution.py`, `hypertrade/paper/relay_netting.py` | 小时级基准相对退化扫描、预算准入、延续账本（遵守 BitPro `relay.blocks_evolution` 跳过接力中的策略；按变体位空位排序、空位已被排队候选预订时让位、AVO 拒绝与 BitPro 已赛过变体完全相同的提案）、赛马裁判只读通知（按变体位逐赛道跟踪、按取消原因区分系统淘汰与人工取消、观察中的赛道慢轮询；BitPro 是唯一接力控制者）、两腿持仓净额平滑换仓（仅限有真实持仓的目标） |
+| **Paper & Evolution** | `hypertrade/arc/evolution.py`, `hypertrade/paper/relay_netting.py` | 小时级基准相对退化扫描、预算准入、延续账本（遵守 BitPro `relay.blocks_evolution` 跳过接力中的策略；按变体位空位排序、空位已被排队候选预订时让位、AVO 拒绝与 BitPro 已赛过变体完全相同的提案）、赛马裁判只读通知（按变体位逐赛道跟踪、按取消原因区分系统淘汰与人工取消、观察中的赛道慢轮询；BitPro 是唯一接力控制者）、赛道结束时记入前向结果表且结果回流研究记忆、两腿持仓净额平滑换仓（仅限有真实持仓的目标） |
| **Target Adapters** | `hypertrade/targets/bitpro.py`, `hypertrade/targets/quantlab.py` | 实现 `market_target.v1` 档案与 `market-evolution.v1` 契约，提供标准化行情、回测、部署与心跳探测 |
-| **Data & Storage** | PostgreSQL 14+, pgvector, 49 项 Alembic 迁移 | 保存规范 Mission 投影、策略资产卡片、实验记录、认知记忆表及飞书告警台账 |
+| **Data & Storage** | PostgreSQL 14+, pgvector, 50 项 Alembic 迁移 | 保存规范 Mission 投影、策略资产卡片、实验记录、认知记忆表及飞书告警台账 |
| **Safety & Ops** | UDS 沙箱, GitHub Actions, Docker Compose, `./scripts/check.sh` | 进程资源限制、网络隔离、静态代码检查（Ruff/Mypy）与 2100+ 项全量自动化测试 |
---
@@ -155,7 +155,7 @@ flowchart TB
## 7. 部署模型与质量门禁
-1. **部署拓扑**：Docker Compose / K8s + Nginx + PostgreSQL (pgvector)。所有服务启动前自动执行 49 项 Alembic 迁移。
+1. **部署拓扑**：Docker Compose / K8s + Nginx + PostgreSQL (pgvector)。所有服务启动前自动执行 50 项 Alembic 迁移。
2. **质量门禁**：通过 `./scripts/check.sh` 实行一票否决制：
- 前端代码检查与 Vitest 测试全绿；
- 前端 Vite 生产构建成功；
diff --git a/docs/contracts/user-directed-evolution-hardening.md b/docs/contracts/user-directed-evolution-hardening.md
index be4aab7..e7a307f 100644
--- a/docs/contracts/user-directed-evolution-hardening.md
+++ b/docs/contracts/user-directed-evolution-hardening.md
@@ -165,9 +165,9 @@ BitPro 的母体现在可以同时让多个挑战者各占一个变体位赛马
- **逐赛道跟踪**：每个 (母体, 挑战者) 一条 `RacePairRecord`。新增可选字段：`slot_no`（位号）、`gate_alpha`（该位的符号检验水平，三个位时约 0.0167）、`plan_seen`、`cancellation_kind` / `cancellation_actor` / `cancellation_reason` / `cancellation_cooldown_hours` / `cancellation_metrics`、`next_poll_at`；缺这些字段的旧状态文件照常加载。卡片显示「位N」，p 值阈值与「及格线：N 个非平局日里至少赢 K 天」按 `gate_alpha` 换算；`gate_alpha` 缺失（旧 BitPro、QuantLab）时按 0.05 且不显示及格线。
- **取消原因**：读取 `relay-control` / `relay-plans` 的 `cancellation.kind`。`operator` 沿用「已取消，操作者终止赛马」；`futility`／`exam`／`timeout` 分别触发里程碑 `FUTILITY`／`EXAM_FAILED`／`TIMED_OUT`（橙色卡片），写明是系统淘汰而非操作者取消、BitPro 给出的原因与证据、释放了哪个变体位，无望淘汰另写冷却小时数，被淘汰的候选不会被自动放行；`superseded`（同母体其他变体已接班）只记 `SUPERSEDED` 标记、不发卡，赢家的 ADOPTED / COMPLETED 卡片已经说明；缺失、格式异常或未知的 kind 一律使用不指名取消者的中性文案（不写「操作者」，也不写「淘汰」）。每个赛道只发一张取消类卡片，旧标记 `CANCELLED` / `REJECTED` 视为已发过；挑战者被重新放行进入新的赛马后清除取消标记，下一次取消可再通知。
-- **消失的赛道**：BitPro 在位置被新挑战者接管后会删除被取消的计划行。曾见过计划而后来查不到的赛道记为 `ended`：保留最后一次证据、不再轮询、不发卡（结束原因 BitPro 自己已推送）；从未见过计划的赛道（如历史文件里已部署但尚未入位的候选）不会被误记为结束。
+- **消失的赛道**：BitPro 在位置被新挑战者接管后会删除被取消的计划行。计划消失时若 `relay-control` 带回淘汰归档（BitPro 已把位置补给下一个候选），按淘汰处理：照常发对应卡片、记输赢，数字保留上次镜像的（不用候选重评的）；不带归档时（旧 BitPro）才记为 `ended`：保留最后一次证据、不再轮询、不发卡（结束原因 BitPro 自己已推送）。从未见过计划的赛道（如历史文件里已部署但尚未入位的候选）不会被误记为结束。
- **轮询降频**：读取一个观察中的赛道会让 BitPro 重载14天小时权益，而赛道数可达母体数×变体位数，因此只在积累证据的赛道每 `RACE_JUDGE_OBSERVING_INTERVAL_SECONDS`（默认900秒，下限60）读一次；交接中（draining / transferred）、即将达标或待宣布的赛道仍按 `RACE_JUDGE_INTERVAL_SECONDS`（默认60秒）每轮查看；`relay-plans` 列表（每轮一次廉价读取）里已跟踪赛道变成 `cancelled` 时立即读取，不等慢轮询；运营端 `POST /api/portfolio/race-judge/run-now` 忽略降频。
-- **重复通知**：一次系统淘汰用户仍会收到两条飞书：BitPro 自己的橙色「赛马淘汰」卡和本卡；是否抑制其一由用户决定，本次没有抑制。
+- **重复通知**：一次系统淘汰用户仍会收到两条飞书：BitPro 自己的橙色「赛马淘汰」卡和本卡；是否抑制其一由用户决定，本次没有抑制。位置被补走的系统淘汰现在也会收到裁判卡片，与其他系统淘汰一致。
验证：`tests/test_race_judge.py` 新增 14 项（各取消原因的文案与里程碑、被接班的赛道不发卡、未知来源中性文案、旧状态文件、旧标记不重发、位号与及格线显示、消失赛道结束、慢轮询与立即读取、重新放行后再次通知、QuantLab 目标无该字段），对 11 处关键守卫做过变异检验，去掉任一守卫都有用例失败。
@@ -181,3 +181,20 @@ BitPro 的母体现在默认有三个变体位（见 BitPro 接力合同「变
- **需求可见**：每轮扫描结果 `payload.relay_seat_demand` 汇总（报告空位的母体数、位总数、已占用、空位、排队候选、`seats_open`），并给出今日剩余研究轮数与每轮最多候选数。若 `seats_open` 长期大于「剩余轮数 × 每轮实际入位的候选数」，说明每日研究预算（`max_research_per_day`，默认4，是有意的成本上限）限制了进化速度；每一轮最多消耗 `max_model_calls` 次模型调用与 `max_backtests` 次回测。生产配置由用户或管理员调整，代码不改默认预算。
验证：`tests/test_arc_relay_seats.py`（新增，空位与重复判定的单元测试）、`tests/test_arc_evolution.py` 新增 5 项（位被排队候选占满即跳过、有空位时研究并把空位事实带入任务、旧 BitPro 行为不变、空位多的母体先研究、被预订的母体让位）、`tests/test_source_variant_avo.py` 新增 2 项（已赛过／排队中／已淘汰的变体在任何副作用前被拒绝，提示词包含该规则）；对9处关键守卫做过变异检验，去掉任一守卫都有用例失败。
+
+## 赛马前向结果回流研究记忆（2026-10-21）
+
+赛马结束的结论此前只停留在裁判状态文件里，研究侧看不到「自己提的改动实际赛成什么样」。现在把已结束赛道写进 `arc_forward_outcomes`，并在研究时回流为有界记忆。
+
+- **数据来源**：赛马裁判在赛道变为 completed／cancelled／ended 时写入表 `arc_forward_outcomes`（每个 目标／母体／挑战者 一行，同一组合只留一行）；进程启动首次巡检时补录状态文件里已结束的记录，并批量对接研究血缘。
+- **结论映射**：completed → `won`；cancelled 且 `kind=futility` → `lost_futility`、`exam` → `lost_exam`、`timeout` → `expired_timeout`、`superseded` → `superseded`、`operator` → `cancelled_operator`、其他 → `cancelled_unknown`；计划消失 → `ended_unknown`。只有 `won`／`lost_futility`／`lost_exam` 是有效输赢证据，其余只记录、不当负面证据。`cancelled_unknown` 与 `ended_unknown` 可被更具体的结论覆盖，其余结论不改写。同一挑战者被重新放行后的第二场结果不会覆盖第一场（已知限制）。
+- **位置被补走**：BitPro 淘汰后可能在裁判下一次扫描前就把位置补给排队候选、删掉旧计划行；BitPro 同期 PR 让 `relay-control` 对已不占位的挑战者附上最近一次淘汰归档，裁判据此按淘汰处理（照常发卡片、记输赢，数字保留上次镜像的、不用候选重评的）；字段缺失（旧 BitPro）时仍记为已结束。
+- **研究血缘**：按挑战者的 BitPro 策略编号找回当初的研究任务与候选（参数与结构改动、作用范围）：先按 `candidate_proposed`／`bitpro_self_tested`／`candidate_validated` 事件定位任务（研究提速到每天最多 32 轮后，按最近任务扫描会让旧任务滑出窗口），再回退扫描最近 500 个任务（覆盖事件内联存储的早期任务）；找不到记为 `no_lineage`，不推断。
+- **回流**：研究某个策略时沿接班链往回最多 3 代（赢家成为新冠军后编号会变，它赢下的那场和前任们的比赛都记在前任名下），取已对接血缘的有效证据最多 8 条（按结束时间倒序），作为 `type=forward_outcome` 的记忆条目放在开发记忆之前（模型只看前 20 条记忆），每条带 `relation`（`made_current`／`variant_of_current`／`made_ancestor`／`variant_of_ancestor`）与 `generations_back`；`memory_id` 为 `outcome:`，可在 `evidence_refs` 中引用；manifest 记录 `forward_outcomes` 计数；读取失败记运行错误、不阻断研究。
+- **提示词规则**：改动是相对那场比赛的母体而言；当前策略的输家方向无新证据与 `repeat_reason` 不重复；赢的方向附近优先；单场结果是噪声观察，不是因果结论，不替代开发证据。
+- **记忆名额**：结果条目最多占模型可见的 20 条记忆中的 8 条，开发记忆的可见条数相应减少（假设评估引用的也是这前 20 条）；这是有意的取舍，前向实测比更早的开发回测更接近真实表现。
+- **可观察**：`GET /api/v1/arc/evolution` 的响应新增 `forward_outcomes`（总数、按结论、按血缘、有效证据数、最近记账时间；读取失败只返回 `error`，不影响其余状态）；研究证据视图 `research.evolution` 新增 `forward_outcome_references`，`memory_references` 只列开发参照。
+- **配套**：BitPro https://github.com/Shadowell/BitPro/pull/1288（relay-control 对已不占位的挑战者附上最近一次淘汰归档），两边部署顺序不限。
+- **未做**：输家逐项消融（第二期）。
+
+验证：`tests/test_forward_outcomes_ledger.py` 覆盖结论映射与证据分级、未知结束升级与具体结论不改写、并发插入对齐、指标未知保持 `None`、血缘定位（事件索引命中滑出窗口的旧任务、重复编号取最新、回退任务扫描、异常载荷不报错）、接班链回溯与排序限量、`summary` 聚合；`tests/test_arc_forward_outcomes_postgres.py` 验证真库迁移升级／降级、唯一组合写入与 JSON 路径查询；`tests/test_race_judge.py` 覆盖赛道结束记账、位置被补走时按淘汰归档发卡并记输赢、归档缺失时仍记结束、重新放行与补录；`tests/test_arc_evolution.py` 覆盖沿接班链回流、按结束时间排序限量、账本失败不阻断研究、结果条目排在开发记忆之前，并断言 `status()` 不含该统计；`tests/test_avo_evidence.py` 与 `tests/test_source_variant_avo.py` 覆盖证据视图只把开发记忆列为参照、`outcome:` 引用可被接受而缺失引用被拒绝、系统提示包含使用规则；`tests/test_arc_external_surface.py` 覆盖 `GET /api/v1/arc/evolution` 的 `forward_outcomes` 计数与账本读取失败时仍返回 200 且带 `error`。
diff --git a/docs/progress.md b/docs/progress.md
index 4f795e1..3e85e9b 100644
--- a/docs/progress.md
+++ b/docs/progress.md
@@ -7,6 +7,13 @@
- **影响分析与风险**：GitNexus 把 `_negative_experiment_finished` 评为 HIGH（唯一直接调用方 `_feedback_child_active`，经预算统计 `usage`/`budget_status`/`admit` 与 `check_paper_feedback`，牵涉进化扫描 `tick`、模拟盘反馈 `check_paper_feedback`、研究观察 `observe_arc_missions_once` 三条流程）；改动只放宽“终验已用掉、数字全过、仅因去偏夏普概率不足被否”这一个窄条件，其余输入判定不变。反馈子任务被去偏夏普否决后同样按“次日再检查”规则释放，不再永久卡住父任务的调优槽位。
- **部署后预期与未做**：合并部署后 509、513 不再 `source_active`，下一轮扫描（每 30 分钟一轮、每轮最多创建一个研究）依次为它们创建研究。333 是挂起的未确认副作用（BitPro 回“幂等键同键不同请求”），需经审计对账工具另行处理，本改动不涉及；443 冷却到 10-22 02:15Z 自然恢复。
+## 2026-10-21：赛马前向结果回流研究记忆
+
+- **背景**：赛马结束的结论此前只留在裁判状态文件里，研究侧看不到自己提的改动实际赛成什么样；研究提速到每天最多 32 轮后，旧任务会滑出「最近 500 个任务」的扫描窗口，血缘更容易对不上。
+- **交付**：迁移 0050 新增 `arc_forward_outcomes`（每个 目标／母体／挑战者 一行）；裁判在完成、取消、结束时记账，位置被补走时按 BitPro 淘汰归档处理，启动时补录已结束赛道；只有 `won`／`lost_futility`／`lost_exam` 是输赢证据，其余只记录不当负面证据；按 BitPro 策略编号经事件索引找回研究血缘（失败再回退任务扫描），对不上记 `no_lineage`；研究时沿接班链回溯最多 3 代、取最多 8 条有效证据作为 `type=forward_outcome` 记忆放在开发记忆之前，manifest 记计数，读取失败不阻断研究；系统提示写明改动相对那场比赛的母体、输家方向无新证据不重复、赢的方向附近优先、单场是噪声不是因果；`GET /api/v1/arc/evolution` 新增 `forward_outcomes` 汇总且失败只带 `error`，研究证据视图新增 `forward_outcome_references`。
+- **验证**：新增 `tests/test_forward_outcomes_ledger.py`、`tests/test_arc_forward_outcomes.py`、`tests/test_arc_forward_outcomes_postgres.py`（PostgreSQL 真库），并在 `test_race_judge`／`test_arc_evolution`／`test_source_variant_avo`／`test_avo_evidence`／`test_arc_external_surface` 中补充用例；对裁判记账、记忆排序与过滤、血缘事件定位等 12 处关键守卫做过变异检验，去掉任一守卫都有用例失败；合入最新 main 后整套 `./scripts/check.sh` 通过（前端 15 项、ruff、mypy、2277 项后端测试）。GitNexus：`EvolutionService._memory` 与血缘查找的影响分析为 HIGH（研究主循环 `tick` 与模拟盘反馈 `check_paper_feedback`），改动为加法、出错降级；`race_judge.py` 的索引映射不可靠（报告为 critical，但所列流程并不调用裁判），已用文本检索核对真实调用方只有 worker 循环与运营端立即运行。未线上实测。
+- **未做**：输家逐项消融为第二期；未线上实测。
+
## 2026-10-21：发布《企业级全自主 Agent 架构设计、核心技术剖析与面试通关指南》（Markdown + 25页 PDF）
- **背景**：针对高阶 Agent 架构学习与大模型算法/工程面试需求，以 HyperTrade 生产系统（11.7 万行 Python 代码、2100+ 测试）为真实解构母本，输出深度架构拆解与实战通关手册。
diff --git a/docs/spec.md b/docs/spec.md
index 0cbd52c..0fd767d 100644
--- a/docs/spec.md
+++ b/docs/spec.md
@@ -1148,7 +1148,7 @@ research start/continue默认持续跟进服务端持久事件；--detach明确
### 每小时自主进化
-BitPro自主进化开关对应HyperTrade持久配置与worker调度。每小时诊断模拟盘7+7退化、当前会话历史成交样本及可追溯开发回测记忆，自主生成改进假设；经过原版本同窗比较和人工审核后，独立运行新Paper。预算、范围、冷却和去重由确定性服务控制。数据不足和不支持复现的组合/策略应显示诊断缺口，不自动替换原策略。默认关闭，立即诊断不会创建研究。（2026-09-14 起 `EvolutionConfig.enabled` 默认开启，仅影响新建配置；见文末「可插拔市场目标与自进化通用化」。）
+BitPro自主进化开关对应HyperTrade持久配置与worker调度。每小时诊断模拟盘7+7退化、当前会话历史成交样本及可追溯开发回测记忆，自主生成改进假设；经过原版本同窗比较和人工审核后，独立运行新Paper。预算、范围、冷却和去重由确定性服务控制。数据不足和不支持复现的组合/策略应显示诊断缺口，不自动替换原策略。赛马赛道结束时把结论写入前向结果表，并沿接班链作为研究记忆回流后续研究。默认关闭，立即诊断不会创建研究。（2026-09-14 起 `EvolutionConfig.enabled` 默认开启，仅影响新建配置；见文末「可插拔市场目标与自进化通用化」。）
### 自主进化的证据上下文与提案合同（2026-09-13）
diff --git a/tests/test_arc_evolution.py b/tests/test_arc_evolution.py
index c291cf9..14391d9 100644
--- a/tests/test_arc_evolution.py
+++ b/tests/test_arc_evolution.py
@@ -17,6 +17,8 @@ def test_defaults_are_hourly_and_enabled(service):
state = service.status()
assert state["config"]["enabled"] is True
assert state["config"]["target_id"] == "bitpro"
+ # The scan calls status() repeatedly; the ledger aggregate belongs to the page only.
+ assert "forward_outcomes" not in state
assert state["config"]["interval_minutes"] == 60
assert state["config"]["threshold_pp"] == "10"
service.configure(EvolutionConfig(enabled=False), revision=0, actor="test")
@@ -806,3 +808,178 @@ def relay(strategy_id):
assert result["payload"]["source_strategy_id"] == 45
held = next(d for d in result["payload"]["diagnostics"] if d["strategy_id"] == 44)
assert held["skip_reason"] == "seats_reserved" and held["relay_seats"]["open"] == 0
+
+
+def _forward_outcome(
+ db,
+ *,
+ parent: str,
+ challenger: str,
+ verdict: str,
+ ended_at: str,
+ linked: bool = True,
+):
+ """Record one finished race; ARC lineage is stated explicitly instead of inferred."""
+ from datetime import datetime
+
+ from hypertrade.arc.evolution_models import ArcForwardOutcome
+ from hypertrade.arc.forward_outcomes import ForwardOutcomeLedger
+
+ row = ForwardOutcomeLedger(db).record(
+ target_id="bitpro",
+ parent_strategy_id=parent,
+ challenger_strategy_id=challenger,
+ verdict=verdict,
+ metrics={"excess_return_pct": 1.5},
+ ended_at=datetime.fromisoformat(ended_at),
+ )
+ if linked:
+ with db.session() as session:
+ stored = session.get(ArcForwardOutcome, row["id"])
+ assert stored is not None
+ stored.lineage_status = "linked"
+ stored.changes_json = {
+ "parameter_changes": {"fast_window": 20},
+ "structural_changes": {},
+ }
+ return row
+
+
+def test_memory_prepends_forward_outcomes_from_the_champion_chain(service, monkeypatch):
+ from hypertrade.arc.store import get_controller
+
+ # 44 won its way into the champion seat it is being evolved from.
+ _forward_outcome(
+ service.db, parent="43", challenger="44", verdict="won", ended_at="2026-01-01T00:00:00"
+ )
+ _forward_outcome(
+ service.db,
+ parent="43",
+ challenger="4301",
+ verdict="lost_exam",
+ ended_at="2026-01-02T00:00:00",
+ )
+ _forward_outcome(
+ service.db,
+ parent="44",
+ challenger="4401",
+ verdict="lost_futility",
+ ended_at="2026-01-03T00:00:00",
+ )
+ # Other lineage, neutral ending and unlinked loss: never memory for this strategy.
+ _forward_outcome(
+ service.db, parent="99", challenger="9901", verdict="won", ended_at="2026-01-04T00:00:00"
+ )
+ _forward_outcome(
+ service.db,
+ parent="44",
+ challenger="4402",
+ verdict="superseded",
+ ended_at="2026-01-05T00:00:00",
+ )
+ _forward_outcome(
+ service.db,
+ parent="44",
+ challenger="4403",
+ verdict="lost_exam",
+ ended_at="2026-01-06T00:00:00",
+ linked=False,
+ )
+
+ now = _variant_service(service, monkeypatch, 100)
+ result = service.tick(now)
+ assert result["status"] == "research_created"
+ context = get_controller(result["payload"]["mission_id"]).projection.goal.evolution_context
+ memory = context["memory"]
+ assert [entry["challenger_strategy_id"] for entry in memory] == ["4401", "4301", "44"]
+ assert [entry["memory_id"].split(":")[0] for entry in memory] == ["outcome"] * 3
+ assert [
+ (entry["relation"], entry["generations_back"], entry["verdict"]) for entry in memory
+ ] == [
+ ("variant_of_current", 0, "lost_futility"),
+ ("variant_of_ancestor", 1, "lost_exam"),
+ ("made_current", 1, "won"),
+ ]
+ assert [entry["type"] for entry in memory] == ["forward_outcome"] * 3
+ assert context["memory_manifest"]["forward_outcomes"] == {
+ "policy": "forward_outcome_v1",
+ "selected": 3,
+ "won": 1,
+ "lost": 2,
+ }
+
+
+def test_memory_continues_when_forward_outcomes_fail(service, monkeypatch):
+ from hypertrade.arc.forward_outcomes import ForwardOutcomeLedger
+ from hypertrade.arc.store import get_controller
+
+ def unavailable(self, **kwargs):
+ raise RuntimeError("ledger unavailable")
+
+ monkeypatch.setattr(ForwardOutcomeLedger, "evidence_for_lineage", unavailable)
+ now = _variant_service(service, monkeypatch, 100)
+ result = service.tick(now)
+ assert result["status"] == "research_created"
+ context = get_controller(result["payload"]["mission_id"]).projection.goal.evolution_context
+ assert context["memory"] == []
+ manifest = context["memory_manifest"]["forward_outcomes"]
+ assert manifest["policy"] == "forward_outcome_v1"
+ assert manifest["selected"] == 0
+ assert "ledger unavailable" in manifest["error"]
+
+
+def test_forward_outcomes_precede_development_memory_in_the_model_view(service, monkeypatch):
+ from hypertrade.arc.avo import _evolution_view
+ from hypertrade.arc.store import get_controller
+ from hypertrade.memory.service import MemoryService
+
+ # Enough development memory to exceed the 20-entry model window on its own.
+ dev = [
+ {
+ "memory_id": f"dev-{i}",
+ "mission_id": f"m-{i}",
+ "candidate_id": f"c-{i}",
+ }
+ for i in range(25)
+ ]
+
+ def project(self, records, **kwargs):
+ return dev, {"policy": "test"}
+
+ monkeypatch.setattr(MemoryService, "project_research", project)
+ # 44 won its way into the champion seat; 4401 raced under it and lost.
+ _forward_outcome(
+ service.db,
+ parent="44",
+ challenger="4401",
+ verdict="lost_futility",
+ ended_at="2026-01-03T00:00:00",
+ )
+ _forward_outcome(
+ service.db, parent="43", challenger="44", verdict="won", ended_at="2026-01-01T00:00:00"
+ )
+
+ now = _variant_service(service, monkeypatch, 100)
+ result = service.tick(now)
+ assert result["status"] == "research_created"
+ context = get_controller(result["payload"]["mission_id"]).projection.goal.evolution_context
+ memory = context["memory"]
+ assert len(memory) == 27
+ # Newest race first; the two outcomes stay ahead of all development entries.
+ assert [entry["type"] for entry in memory[:2]] == ["forward_outcome"] * 2
+ assert [entry["challenger_strategy_id"] for entry in memory[:2]] == ["4401", "44"]
+ assert [entry["memory_id"] for entry in memory[2:]] == [f"dev-{i}" for i in range(25)]
+ assert context["memory_manifest"]["policy"] == "test"
+ assert context["memory_manifest"]["forward_outcomes"] == {
+ "policy": "forward_outcome_v1",
+ "selected": 2,
+ "won": 1,
+ "lost": 1,
+ }
+
+ # The model only reads the first 20 entries, so ordering decides what it sees.
+ view = _evolution_view(context)
+ assert len(view["memory"]) == 20
+ assert [entry["type"] for entry in view["memory"][:2]] == ["forward_outcome"] * 2
+ assert view["memory"][2]["memory_id"] == "dev-0"
+ assert view["memory_in_view"] == {"shown": 20, "total": 27}
diff --git a/tests/test_arc_external_surface.py b/tests/test_arc_external_surface.py
index 0c023ad..935f1bd 100644
--- a/tests/test_arc_external_surface.py
+++ b/tests/test_arc_external_surface.py
@@ -588,3 +588,43 @@ def test_research_token_cannot_enable_automatic_paper_policy(client):
).status_code
== 403
)
+
+
+def _record_forward_outcome(client: TestClient, challenger: int, verdict: str) -> None:
+ from hypertrade.arc.forward_outcomes import ForwardOutcomeLedger
+
+ ForwardOutcomeLedger(client.app.state.db).record(
+ target_id="bitpro",
+ parent_strategy_id=107,
+ challenger_strategy_id=challenger,
+ verdict=verdict,
+ metrics={},
+ link=False,
+ )
+
+
+def test_evolution_status_carries_forward_outcome_summary(client: TestClient) -> None:
+ _record_forward_outcome(client, 9001, "won")
+ _record_forward_outcome(client, 9002, "superseded")
+
+ response = client.get("/api/v1/arc/evolution", headers=_headers(READ_TOKEN))
+
+ assert response.status_code == 200
+ outcomes = response.json()["forward_outcomes"]
+ assert outcomes["total"] == 2
+ assert outcomes["evidence"] == 1
+ assert outcomes["by_verdict"] == {"superseded": 1, "won": 1}
+
+
+def test_evolution_status_survives_forward_outcome_failure(client: TestClient, monkeypatch) -> None:
+ from hypertrade.arc.forward_outcomes import ForwardOutcomeLedger
+
+ def fail(_self) -> dict:
+ raise RuntimeError("ledger unavailable")
+
+ monkeypatch.setattr(ForwardOutcomeLedger, "summary", fail)
+
+ response = client.get("/api/v1/arc/evolution", headers=_headers(READ_TOKEN))
+
+ assert response.status_code == 200
+ assert "error" in response.json()["forward_outcomes"]
diff --git a/tests/test_arc_forward_outcomes.py b/tests/test_arc_forward_outcomes.py
new file mode 100644
index 0000000..b712f26
--- /dev/null
+++ b/tests/test_arc_forward_outcomes.py
@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+
+import pytest
+from hypertrade.arc.evolution_models import ArcForwardOutcome
+from hypertrade.db import Database
+from sqlalchemy import create_engine, inspect
+from sqlalchemy.exc import IntegrityError
+
+COLUMNS = {
+ "id",
+ "target_id",
+ "parent_strategy_id",
+ "challenger_strategy_id",
+ "slot_no",
+ "generation",
+ "verdict",
+ "verdict_source",
+ "ended_at",
+ "metrics_json",
+ "evidence_sha256",
+ "lineage_status",
+ "mission_id",
+ "candidate_id",
+ "attempt_id",
+ "changes_json",
+ "scope_json",
+ "created_at",
+ "updated_at",
+}
+INDEXES = {
+ "ix_arc_forward_outcomes_parent_strategy_id",
+ "ix_arc_forward_outcomes_challenger_strategy_id",
+ "ix_arc_forward_outcomes_verdict",
+ "ix_arc_forward_outcomes_lineage_status",
+ "ix_arc_forward_outcomes_mission_id",
+}
+
+
+def _migration():
+ path = (
+ Path(__file__).parents[1] / "backend/alembic/versions/0050_arc_forward_outcomes.py"
+ )
+ spec = importlib.util.spec_from_file_location("arc_forward_outcomes_migration", path)
+ assert spec is not None and spec.loader is not None
+ module = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(module)
+ return module
+
+
+def test_forward_outcome_migration_creates_and_removes_only_its_table(tmp_path: Path) -> None:
+ from alembic.migration import MigrationContext
+ from alembic.operations import Operations
+
+ migration = _migration()
+ engine = create_engine(f"sqlite:///{tmp_path / 'migration.db'}")
+ with engine.begin() as connection, Operations.context(MigrationContext.configure(connection)):
+ migration.upgrade()
+ inspector = inspect(connection)
+ assert inspector.get_table_names() == ["arc_forward_outcomes"]
+ assert {column["name"] for column in inspector.get_columns(
+ "arc_forward_outcomes"
+ )} == COLUMNS
+ assert inspector.get_pk_constraint("arc_forward_outcomes")["constrained_columns"] == [
+ "id"
+ ]
+ uniques = {
+ item["name"]
+ for item in inspector.get_unique_constraints("arc_forward_outcomes")
+ }
+ assert "uq_arc_forward_outcome_pair" in uniques
+ indexes = {
+ item["name"] for item in inspector.get_indexes("arc_forward_outcomes")
+ }
+ assert INDEXES.issubset(indexes)
+
+ migration.downgrade()
+ assert inspect(connection).get_table_names() == []
+ engine.dispose()
+
+
+def test_forward_outcome_round_trips_json_and_leaves_unknown_optionals_empty() -> None:
+ db = Database("sqlite:///:memory:")
+ db.create_all()
+
+ metrics = {"relative_pnl": 0.08, "ranking": {"rank": 2, "of": 3}, "note": None}
+ changes = {"fast_window": 7}
+ scope = {"symbols": ["BTC-USDT-SWAP"], "timeframe": "1H"}
+ with db.session() as session:
+ session.add(
+ ArcForwardOutcome(
+ id="fo_" + "a" * 32,
+ target_id="bitpro",
+ parent_strategy_id="41",
+ challenger_strategy_id="42",
+ verdict="handoff",
+ metrics_json=metrics,
+ changes_json=changes,
+ scope_json=scope,
+ )
+ )
+
+ with db.session() as session:
+ row = session.get(ArcForwardOutcome, "fo_" + "a" * 32)
+ assert row is not None
+ assert row.metrics_json == metrics
+ assert row.changes_json == changes
+ assert row.scope_json == scope
+ assert row.generation == 1
+ assert row.verdict_source == "race_judge"
+ assert row.lineage_status == "pending"
+ assert row.evidence_sha256 == ""
+ assert row.slot_no is None
+ assert row.ended_at is None
+ assert row.mission_id is None
+ assert row.candidate_id is None
+ assert row.attempt_id is None
+
+
+def test_forward_outcome_duplicate_pair_is_rejected() -> None:
+ db = Database("sqlite:///:memory:")
+ db.create_all()
+ pair = {
+ "target_id": "bitpro",
+ "parent_strategy_id": "41",
+ "challenger_strategy_id": "42",
+ "verdict": "handoff",
+ }
+
+ with db.session() as session:
+ session.add(ArcForwardOutcome(id="fo_" + "a" * 32, **pair))
+
+ with pytest.raises(IntegrityError), db.session() as session:
+ session.add(ArcForwardOutcome(id="fo_" + "b" * 32, **pair))
diff --git a/tests/test_arc_forward_outcomes_postgres.py b/tests/test_arc_forward_outcomes_postgres.py
new file mode 100644
index 0000000..9058ae8
--- /dev/null
+++ b/tests/test_arc_forward_outcomes_postgres.py
@@ -0,0 +1,123 @@
+"""Forward-outcome migration and unique pair on an isolated PostgreSQL cluster."""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy import inspect, text
+from sqlalchemy.exc import IntegrityError
+from test_arc_event_journal_postgres import _alembic, _cluster
+
+INSERT = text(
+ "INSERT INTO arc_forward_outcomes"
+ " (id, target_id, parent_strategy_id, challenger_strategy_id, verdict,"
+ " metrics_json, changes_json, scope_json)"
+ " VALUES (:id, :target, :parent, :challenger, :verdict,"
+ " CAST(:metrics AS json), CAST(:changes AS json), CAST(:scope AS json))"
+)
+
+
+def test_postgres_upgrade_write_unique_pair_and_downgrade(tmp_path) -> None:
+ with _cluster(tmp_path) as url:
+ from hypertrade.db import Database
+
+ db = Database(url)
+ with db.engine.begin() as connection:
+ connection.execute(
+ text("CREATE TABLE alembic_version (version_num VARCHAR(32) PRIMARY KEY)")
+ )
+ connection.execute(
+ text("INSERT INTO alembic_version VALUES ('0049_arc_cognitive_memory')")
+ )
+
+ _alembic(url, "upgrade", "0050_arc_forward_outcomes")
+
+ row = {
+ "id": "fo_" + "a" * 32,
+ "target": "bitpro",
+ "parent": "41",
+ "challenger": "42",
+ "verdict": "handoff",
+ "metrics": '{"relative_pnl": 0.08}',
+ "changes": '{"fast_window": 7}',
+ "scope": '{"symbols": ["BTC-USDT-SWAP"]}',
+ }
+ with db.engine.begin() as connection:
+ connection.execute(INSERT, row)
+
+ with db.engine.connect() as connection:
+ stored = connection.execute(
+ text(
+ "SELECT generation, verdict_source, lineage_status, evidence_sha256,"
+ " slot_no, mission_id, metrics_json, changes_json, scope_json"
+ " FROM arc_forward_outcomes WHERE id = :id"
+ ),
+ {"id": row["id"]},
+ ).mappings().one()
+ assert stored["generation"] == 1
+ assert stored["verdict_source"] == "race_judge"
+ assert stored["lineage_status"] == "pending"
+ assert stored["evidence_sha256"] == ""
+ assert stored["slot_no"] is None
+ assert stored["mission_id"] is None
+ assert stored["metrics_json"] == {"relative_pnl": 0.08}
+ assert stored["changes_json"] == {"fast_window": 7}
+ assert stored["scope_json"] == {"symbols": ["BTC-USDT-SWAP"]}
+
+ with pytest.raises(IntegrityError), db.engine.begin() as connection:
+ connection.execute(INSERT, {**row, "id": "fo_" + "b" * 32})
+
+ db.engine.dispose()
+ _alembic(url, "downgrade", "0049_arc_cognitive_memory")
+ downgraded = Database(url)
+ with downgraded.engine.connect() as connection:
+ version = connection.execute(
+ text("SELECT version_num FROM alembic_version")
+ ).scalar_one()
+ assert version == "0049_arc_cognitive_memory"
+ assert "arc_forward_outcomes" not in inspect(connection).get_table_names()
+ downgraded.engine.dispose()
+
+
+def test_postgres_lineage_event_index_normalizes_json_ids(tmp_path) -> None:
+ with _cluster(tmp_path) as url:
+ from hypertrade.arc.forward_outcomes import ForwardOutcomeLedger
+ from hypertrade.db import ArcMissionEvent, Database
+
+ db = Database(url)
+ db.create_all()
+ events = [
+ ("m-old", "evt_old", "bitpro_self_tested", {"payload": {"bitpro_strategy_id": "9501"}}),
+ ("m-new", "evt_new", "candidate_validated", {"payload": {"bitpro_strategy_id": 9501}}),
+ (
+ "m-prop",
+ "evt_prop",
+ "candidate_proposed",
+ {"payload": {"attempt": {"bitpro_strategy_id": "9502"}}},
+ ),
+ ("m-bad", "evt_bad", "bitpro_self_tested", {"payload": "not a dict"}),
+ (
+ "m-ignored",
+ "evt_ignored",
+ "paper_feedback_checked",
+ {"payload": {"bitpro_strategy_id": "9503"}},
+ ),
+ ]
+ with db.session() as session:
+ for seq, (mission_id, event_id, event_type, event_json) in enumerate(
+ events, start=1
+ ):
+ session.add(
+ ArcMissionEvent(
+ mission_id=mission_id,
+ seq=seq,
+ event_id=event_id,
+ event_type=event_type,
+ event_json=event_json,
+ )
+ )
+ session.flush()
+ located = ForwardOutcomeLedger(db)._locate_by_events(
+ session, {"9501", "9502", "9503"}
+ )
+ assert located == {"9501": "m-new", "9502": "m-prop"}
+ db.engine.dispose()
diff --git a/tests/test_avo_evidence.py b/tests/test_avo_evidence.py
index c5a58c8..51557b4 100644
--- a/tests/test_avo_evidence.py
+++ b/tests/test_avo_evidence.py
@@ -30,3 +30,49 @@ def test_development_metrics_stay_separate_from_final_metrics():
final = build_evidence_view(ctrl.projection)
assert final["candidates"][0]["oos_sharpe"] == 1.5
assert final["candidates"][0]["trades"] == 35
+
+
+def test_forward_outcome_memory_is_not_reported_as_development_reference():
+ ctrl = ARCController(
+ goal=ARCGoalV1(
+ objective="test",
+ research_mode="avo",
+ paper_review_required=True,
+ evolution_context={
+ "memory": [
+ {
+ "mission_id": "m_dev",
+ "candidate_id": "c_dev",
+ "development": {"backtest_id": "bt_dev"},
+ },
+ {
+ "memory_id": "outcome:fo_1",
+ "type": "forward_outcome",
+ "mission_id": "m_outcome",
+ "candidate_id": "c_outcome",
+ "verdict": "won",
+ "relation": "made_current",
+ "generations_back": 1,
+ "parent_strategy_id": "43",
+ "challenger_strategy_id": "44",
+ "ended_at": "2026-01-01T00:00:00",
+ },
+ ]
+ },
+ )
+ )
+ evolution = build_evidence_view(ctrl.projection)["research"]["evolution"]
+ assert evolution["memory_references"] == [
+ {"mission_id": "m_dev", "candidate_id": "c_dev", "backtest_id": "bt_dev"}
+ ]
+ assert evolution["forward_outcome_references"] == [
+ {
+ "memory_id": "outcome:fo_1",
+ "verdict": "won",
+ "relation": "made_current",
+ "generations_back": 1,
+ "parent_strategy_id": "43",
+ "challenger_strategy_id": "44",
+ "ended_at": "2026-01-01T00:00:00",
+ }
+ ]
diff --git a/tests/test_forward_outcomes_ledger.py b/tests/test_forward_outcomes_ledger.py
new file mode 100644
index 0000000..0b4ee40
--- /dev/null
+++ b/tests/test_forward_outcomes_ledger.py
@@ -0,0 +1,706 @@
+from __future__ import annotations
+
+from datetime import datetime
+
+import hypertrade.arc.forward_outcomes as forward_outcomes
+import pytest
+from hypertrade.arc.contracts import ARCCandidateAttemptV1, ARCGoalV1
+from hypertrade.arc.controller import ARCController
+from hypertrade.arc.evolution_models import ArcForwardOutcome
+from hypertrade.arc.forward_outcomes import (
+ EVIDENCE_VERDICTS,
+ NEUTRAL_VERDICTS,
+ ForwardOutcomeLedger,
+ metrics_from_record,
+ outcome_id,
+ verdict_for,
+)
+from hypertrade.arc.store import configure_store, reset_store, save_mission
+from hypertrade.db import ArcMission, ArcMissionEvent, Database
+from hypertrade.paper.race_judge import RacePairRecord
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+METRIC_KEYS = {
+ "excess_return_pct",
+ "parent_net_return_pct",
+ "challenger_net_return_pct",
+ "parent_max_drawdown_pct",
+ "challenger_max_drawdown_pct",
+ "parent_trades",
+ "challenger_trades",
+ "observed_hours",
+ "sign_test_p",
+ "gate_alpha",
+ "positive_days",
+ "paired_non_tie_days",
+ "slot_no",
+ "cancellation_reason",
+ "cancellation_metrics",
+}
+
+
+@pytest.fixture
+def ledger():
+ db = Database("sqlite:///:memory:")
+ db.create_all()
+ configure_store(db)
+ yield ForwardOutcomeLedger(db)
+ reset_store()
+
+
+def _save_mission(*challengers: str) -> ARCController:
+ """One ARC mission whose attempts launched the given BitPro challengers."""
+ ctrl = ARCController(goal=ARCGoalV1(objective="race outcomes", symbols=["BTC-USDT"]))
+ for index, challenger in enumerate(challengers):
+ ctrl.projection.attempts.append(
+ ARCCandidateAttemptV1(
+ attempt_id=f"attempt-{challenger}",
+ candidate_id=f"candidate-{challenger}",
+ hypothesis=f"variant {challenger}",
+ strategy_code="class Variant: pass\n",
+ strategy_spec={
+ "parameter_changes": {"fast_window": 10 + index},
+ "structural_changes": {},
+ "symbol": "BTC-USDT",
+ "symbols": ["BTC-USDT"],
+ "timeframe": "1H",
+ "parent_strategy_id": 107,
+ },
+ bitpro_strategy_id=challenger,
+ )
+ )
+ save_mission(ctrl)
+ return ctrl
+
+
+def _race_record(**overrides) -> dict:
+ fields: dict = {
+ "observed_hours": 100,
+ "parent_trades": 11,
+ "challenger_trades": 13,
+ "parent_net_return_pct": 1.0,
+ "challenger_net_return_pct": 2.0,
+ "excess_return_pct": 1.0,
+ "parent_max_drawdown_pct": 3.0,
+ "challenger_max_drawdown_pct": 4.0,
+ "sign_test_p": 0.04,
+ "positive_days": 8,
+ "paired_non_tie_days": 12,
+ "slot_no": 2,
+ "gate_alpha": 0.025,
+ "cancellation_reason": "too slow",
+ "cancellation_metrics": {"common_hours": 100, "excess_pp": -2.0},
+ }
+ fields.update(overrides)
+ return RacePairRecord(
+ parent_strategy_id=107, challenger_strategy_id=9001, **fields
+ ).to_dict()
+
+
+def _record(
+ ledger: ForwardOutcomeLedger,
+ *,
+ challenger: str,
+ verdict: str,
+ parent: int = 107,
+ metrics: dict | None = None,
+ ended_at: datetime | None = None,
+ source: str = "race_judge",
+ link: bool = True,
+) -> dict:
+ return ledger.record(
+ target_id="bitpro",
+ parent_strategy_id=parent,
+ challenger_strategy_id=challenger,
+ verdict=verdict,
+ metrics=metrics or {},
+ ended_at=ended_at,
+ source=source,
+ link=link,
+ )
+
+
+def test_outcome_id_is_stable_and_normalizes_ids() -> None:
+ first = outcome_id("bitpro", 107, 9001)
+ assert first == outcome_id("bitpro", "107", "9001")
+ assert first.startswith("fo_")
+ assert len(first) == 3 + 32
+ assert first != outcome_id("bitpro", 107, 9002)
+
+
+def test_verdict_for_maps_every_finished_state() -> None:
+ assert verdict_for("completed", None) == "won"
+ assert verdict_for("cancelled", "futility") == "lost_futility"
+ assert verdict_for("cancelled", "exam") == "lost_exam"
+ assert verdict_for("cancelled", "timeout") == "expired_timeout"
+ assert verdict_for("cancelled", "superseded") == "superseded"
+ assert verdict_for("cancelled", "operator") == "cancelled_operator"
+ assert verdict_for("cancelled", "system") == "cancelled_unknown"
+ assert verdict_for("cancelled", None) == "cancelled_unknown"
+ assert verdict_for("cancelled", "new_kind_from_the_future") == "cancelled_unknown"
+ assert verdict_for("ended", None) == "ended_unknown"
+ for state in ("observing", "draining", "transferred", "rejected", None):
+ assert verdict_for(state, None) is None
+
+
+def test_verdict_constants_classify_evidence_and_neutral_results() -> None:
+ assert EVIDENCE_VERDICTS == ("won", "lost_futility", "lost_exam")
+ assert set(NEUTRAL_VERDICTS).isdisjoint(EVIDENCE_VERDICTS)
+
+
+def test_metrics_from_record_keeps_known_values() -> None:
+ metrics = metrics_from_record(_race_record())
+ assert set(metrics) == METRIC_KEYS
+ assert metrics["excess_return_pct"] == 1.0
+ assert metrics["parent_net_return_pct"] == 1.0
+ assert metrics["challenger_net_return_pct"] == 2.0
+ assert metrics["parent_trades"] == 11
+ assert metrics["challenger_trades"] == 13
+ assert metrics["observed_hours"] == 100
+ assert metrics["sign_test_p"] == 0.04
+ assert metrics["gate_alpha"] == 0.025
+ assert metrics["positive_days"] == 8
+ assert metrics["paired_non_tie_days"] == 12
+ assert metrics["slot_no"] == 2
+ assert metrics["cancellation_reason"] == "too slow"
+ assert metrics["cancellation_metrics"] == {"common_hours": 100, "excess_pp": -2.0}
+
+
+def test_metrics_from_record_leaves_unknown_as_none() -> None:
+ metrics = metrics_from_record({})
+ assert set(metrics) == METRIC_KEYS
+ assert all(value is None for value in metrics.values())
+
+
+def test_metrics_from_record_treats_missing_or_zero_non_tie_days_as_unknown() -> None:
+ metrics = metrics_from_record(_race_record(positive_days=0, paired_non_tie_days=0))
+ assert metrics["positive_days"] is None
+ assert metrics["paired_non_tie_days"] is None
+
+ record = _race_record()
+ record.pop("paired_non_tie_days")
+ metrics = metrics_from_record(record)
+ assert metrics["positive_days"] is None
+ assert metrics["paired_non_tie_days"] is None
+
+
+def test_metrics_from_record_rejects_booleans_and_non_finite_numbers() -> None:
+ metrics = metrics_from_record(
+ _race_record(
+ excess_return_pct=float("nan"),
+ parent_trades=True,
+ challenger_trades=float("inf"),
+ sign_test_p=float("-inf"),
+ cancellation_metrics="not a dict",
+ )
+ )
+ assert metrics["excess_return_pct"] is None
+ assert metrics["parent_trades"] is None
+ assert metrics["challenger_trades"] is None
+ assert metrics["sign_test_p"] is None
+ assert metrics["cancellation_metrics"] is None
+
+
+def test_record_is_idempotent_and_keeps_first_evidence(ledger: ForwardOutcomeLedger) -> None:
+ first = _record(
+ ledger,
+ challenger="9001",
+ verdict="lost_exam",
+ metrics={"excess_return_pct": -1.0},
+ )
+ second = _record(
+ ledger,
+ challenger="9001",
+ verdict="lost_exam",
+ metrics={"excess_return_pct": -2.0},
+ )
+ assert second["id"] == first["id"]
+ assert second["verdict"] == "lost_exam"
+ assert second["metrics"] == {"excess_return_pct": -1.0}
+ assert second["evidence_sha256"] == first["evidence_sha256"]
+ with ledger.db.session() as session:
+ assert len(session.scalars(select(ArcForwardOutcome)).all()) == 1
+
+
+def test_concrete_verdict_is_never_rewritten(ledger: ForwardOutcomeLedger) -> None:
+ first = _record(
+ ledger,
+ challenger="9001",
+ verdict="lost_exam",
+ metrics={"excess_return_pct": -1.0},
+ )
+ second = _record(
+ ledger,
+ challenger="9001",
+ verdict="won",
+ metrics={"excess_return_pct": 5.0},
+ )
+ assert second["verdict"] == "lost_exam"
+ assert second["metrics"] == first["metrics"]
+ assert second["evidence_sha256"] == first["evidence_sha256"]
+
+
+def test_unknown_ending_can_be_upgraded_to_a_concrete_verdict(
+ ledger: ForwardOutcomeLedger,
+) -> None:
+ _record(ledger, challenger="9001", verdict="ended_unknown")
+ upgraded = _record(
+ ledger,
+ challenger="9001",
+ verdict="lost_futility",
+ metrics={"excess_return_pct": -3.0},
+ ended_at=datetime(2026, 1, 2, 3, 4),
+ source="operator",
+ )
+ assert upgraded["verdict"] == "lost_futility"
+ assert upgraded["verdict_source"] == "operator"
+ assert upgraded["metrics"] == {"excess_return_pct": -3.0}
+ assert upgraded["ended_at"] is not None
+ assert upgraded["ended_at"].startswith("2026-01-02T03:04")
+
+
+def test_one_unknown_verdict_does_not_replace_another(ledger: ForwardOutcomeLedger) -> None:
+ _record(ledger, challenger="9001", verdict="ended_unknown")
+ result = _record(ledger, challenger="9001", verdict="cancelled_unknown")
+ assert result["verdict"] == "ended_unknown"
+
+
+def test_unknown_verdict_is_rejected(ledger: ForwardOutcomeLedger) -> None:
+ with pytest.raises(ValueError):
+ _record(ledger, challenger="9001", verdict="handoff")
+
+
+def test_record_links_matching_mission_attempt(ledger: ForwardOutcomeLedger) -> None:
+ mission = _save_mission("9001")
+ row = ledger.record(
+ target_id="bitpro",
+ parent_strategy_id=107,
+ challenger_strategy_id="9001",
+ verdict="won",
+ metrics={"excess_return_pct": 2.5},
+ ended_at=datetime(2026, 1, 1),
+ slot_no=2,
+ )
+ assert row["lineage_status"] == "linked"
+ assert row["mission_id"] == mission.mission_id
+ assert row["candidate_id"] == "candidate-9001"
+ assert row["attempt_id"] == "attempt-9001"
+ assert row["changes"] == {
+ "parameter_changes": {"fast_window": 10},
+ "structural_changes": {},
+ }
+ assert row["scope"] == {
+ "symbol": "BTC-USDT",
+ "symbols": ["BTC-USDT"],
+ "timeframe": "1H",
+ "parent_strategy_id": 107,
+ }
+ assert row["slot_no"] == 2
+
+
+def test_record_without_matching_attempt_is_no_lineage(
+ ledger: ForwardOutcomeLedger,
+) -> None:
+ _save_mission("9001")
+ row = _record(ledger, challenger="9999", verdict="won")
+ assert row["lineage_status"] == "no_lineage"
+ assert row["mission_id"] is None
+ assert row["changes"] == {}
+ assert row["scope"] == {}
+
+
+def test_link_pending_links_rows_left_pending(ledger: ForwardOutcomeLedger) -> None:
+ mission = _save_mission("9002")
+ oid = outcome_id("bitpro", 107, "9002")
+ with ledger.db.session() as session:
+ session.add(
+ ArcForwardOutcome(
+ id=oid,
+ target_id="bitpro",
+ parent_strategy_id="107",
+ challenger_strategy_id="9002",
+ verdict="won",
+ metrics_json={},
+ lineage_status="pending",
+ )
+ )
+
+ assert ledger.link_pending() == 1
+ with ledger.db.session() as session:
+ row = session.get(ArcForwardOutcome, oid)
+ assert row is not None
+ assert row.lineage_status == "linked"
+ assert row.mission_id == mission.mission_id
+ assert ledger.link_pending() == 0
+
+
+def test_link_pending_matches_a_whole_batch_with_one_scan(
+ ledger: ForwardOutcomeLedger, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ missions = [_save_mission("9101"), _save_mission("9102"), _save_mission("9103")]
+ for challenger in ("9101", "9102", "9103"):
+ row = _record(ledger, challenger=challenger, verdict="won", link=False)
+ assert row["lineage_status"] == "pending"
+
+ loads: list[str] = []
+ real_load = forward_outcomes.load_projection
+
+ def counting_load(session, mission):
+ loads.append(mission.mission_id)
+ return real_load(session, mission)
+
+ monkeypatch.setattr(forward_outcomes, "load_projection", counting_load)
+
+ assert ledger.link_pending() == 3
+ assert len(loads) <= 3
+ with ledger.db.session() as session:
+ rows = {
+ row.challenger_strategy_id: row
+ for row in session.scalars(select(ArcForwardOutcome)).all()
+ }
+ for mission, challenger in zip(missions, ("9101", "9102", "9103"), strict=True):
+ assert rows[challenger].lineage_status == "linked"
+ assert rows[challenger].mission_id == mission.mission_id
+
+
+def test_link_stops_scanning_after_the_newest_match(
+ ledger: ForwardOutcomeLedger, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ missions = [_save_mission("9101"), _save_mission("9102"), _save_mission("9103")]
+ with ledger.db.session() as session:
+ for minutes, mission in enumerate(missions):
+ stored = session.get(ArcMission, mission.mission_id)
+ assert stored is not None
+ stored.updated_at = datetime(2026, 1, 1, 0, minutes)
+
+ row = _record(ledger, challenger="9103", verdict="won", link=False)
+ loads: list[str] = []
+ real_load = forward_outcomes.load_projection
+
+ def counting_load(session, mission):
+ loads.append(mission.mission_id)
+ return real_load(session, mission)
+
+ monkeypatch.setattr(forward_outcomes, "load_projection", counting_load)
+
+ linked = ledger.link(row["id"])
+ assert linked is not None
+ assert linked["mission_id"] == missions[2].mission_id
+ assert loads == [missions[2].mission_id]
+
+
+def test_record_recovers_when_another_writer_inserted_first(
+ ledger: ForwardOutcomeLedger, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ oid = outcome_id("bitpro", 107, "9201")
+ with ledger.db.session() as session:
+ session.add(
+ ArcForwardOutcome(
+ id=oid,
+ target_id="bitpro",
+ parent_strategy_id="107",
+ challenger_strategy_id="9201",
+ verdict="ended_unknown",
+ metrics_json={},
+ evidence_sha256="stale",
+ lineage_status="no_lineage",
+ )
+ )
+
+ real_get = Session.get
+ first = {"pending": True}
+
+ def get_once_none(self, entity, ident, **kwargs):
+ if first["pending"]:
+ first["pending"] = False
+ return None
+ return real_get(self, entity, ident, **kwargs)
+
+ monkeypatch.setattr(Session, "get", get_once_none)
+
+ result = _record(
+ ledger,
+ challenger="9201",
+ verdict="lost_exam",
+ metrics={"excess_return_pct": -0.5},
+ )
+ assert result["verdict"] == "lost_exam"
+ assert result["metrics"] == {"excess_return_pct": -0.5}
+ with ledger.db.session() as session:
+ rows = session.scalars(select(ArcForwardOutcome)).all()
+ assert len(rows) == 1
+ assert rows[0].verdict == "lost_exam"
+
+
+def test_evidence_for_lineage_ancestors_zero_filters_sorts_and_limits(
+ ledger: ForwardOutcomeLedger,
+) -> None:
+ _save_mission("9001", "9002", "9003", "9004", "9006")
+ _record(
+ ledger,
+ challenger="9001",
+ verdict="won",
+ ended_at=datetime(2026, 1, 1),
+ )
+ _record(
+ ledger,
+ challenger="9002",
+ verdict="lost_exam",
+ ended_at=datetime(2026, 1, 3),
+ )
+ _record(
+ ledger,
+ challenger="9003",
+ verdict="expired_timeout",
+ ended_at=datetime(2026, 1, 4),
+ )
+ _record(
+ ledger,
+ challenger="9004",
+ parent=108,
+ verdict="lost_futility",
+ ended_at=datetime(2026, 1, 2),
+ )
+ # Launched outside ARC research: concrete, but not lineage-linked evidence.
+ _record(
+ ledger,
+ challenger="9005",
+ verdict="won",
+ ended_at=datetime(2026, 1, 5),
+ )
+ _record(ledger, challenger="9006", verdict="won", ended_at=None)
+
+ results = ledger.evidence_for_lineage(target_id="bitpro", strategy_id=107, ancestors=0)
+ assert [row["challenger_strategy_id"] for row in results] == ["9002", "9001", "9006"]
+ assert all(row["verdict"] in EVIDENCE_VERDICTS for row in results)
+ assert all(row["lineage_status"] == "linked" for row in results)
+ # 9001 won, but it is not the id under research, so it is still a variant of it.
+ assert all(row["relation"] == "variant_of_current" for row in results)
+ assert all(row["generations_back"] == 0 for row in results)
+
+ limited = ledger.evidence_for_lineage(
+ target_id="bitpro", strategy_id=107, ancestors=0, limit=2
+ )
+ assert [row["challenger_strategy_id"] for row in limited] == ["9002", "9001"]
+
+
+def test_evidence_for_lineage_walks_the_champion_chain(ledger: ForwardOutcomeLedger) -> None:
+ _save_mission("101", "102", "1001", "1011", "1021", "9991", "1012")
+ races = (
+ (100, "1001", "lost_exam", datetime(2026, 1, 1)),
+ (100, "101", "won", datetime(2026, 1, 2)),
+ (101, "1011", "lost_futility", datetime(2026, 1, 3)),
+ (101, "102", "won", datetime(2026, 1, 4)),
+ (102, "1021", "lost_exam", datetime(2026, 1, 5)),
+ # Other lineage, neutral ending and unlinked loss: never evidence here.
+ (999, "9991", "won", datetime(2026, 1, 6)),
+ (101, "1012", "superseded", datetime(2026, 1, 7)),
+ (102, "1022", "lost_exam", datetime(2026, 1, 8)),
+ )
+ for parent, challenger, verdict, ended_at in races:
+ _record(
+ ledger,
+ parent=parent,
+ challenger=challenger,
+ verdict=verdict,
+ ended_at=ended_at,
+ )
+
+ results = ledger.evidence_for_lineage(target_id="bitpro", strategy_id="102", ancestors=3)
+ assert [row["challenger_strategy_id"] for row in results] == [
+ "1021",
+ "102",
+ "1011",
+ "101",
+ "1001",
+ ]
+ assert [
+ (row["challenger_strategy_id"], row["relation"], row["generations_back"])
+ for row in results
+ ] == [
+ ("1021", "variant_of_current", 0),
+ ("102", "made_current", 1),
+ ("1011", "variant_of_ancestor", 1),
+ ("101", "made_ancestor", 2),
+ ("1001", "variant_of_ancestor", 2),
+ ]
+
+ near = ledger.evidence_for_lineage(target_id="bitpro", strategy_id="102", ancestors=1)
+ assert [row["challenger_strategy_id"] for row in near] == ["1021", "102", "1011"]
+
+ # A corrupted won link back to the current champion must not loop forever.
+ _record(
+ ledger,
+ parent=102,
+ challenger="100",
+ verdict="won",
+ ended_at=datetime(2026, 1, 9),
+ )
+ cyclic = ledger.evidence_for_lineage(target_id="bitpro", strategy_id="102", ancestors=10)
+ assert len(cyclic) == len({row["id"] for row in cyclic}) == 5
+
+
+def test_summary_counts_verdicts_lineage_and_evidence(ledger: ForwardOutcomeLedger) -> None:
+ _save_mission("9001", "9002", "9003", "9004")
+ _record(ledger, challenger="9001", verdict="won")
+ _record(ledger, challenger="9002", verdict="lost_futility")
+ _record(ledger, challenger="9003", verdict="expired_timeout")
+ _record(ledger, challenger="9004", verdict="ended_unknown")
+
+ summary = ledger.summary()
+ assert summary["total"] == 4
+ assert summary["by_verdict"] == {
+ "won": 1,
+ "lost_futility": 1,
+ "expired_timeout": 1,
+ "ended_unknown": 1,
+ }
+ assert summary["by_lineage"] == {"linked": 4}
+ assert summary["evidence"] == 2
+ assert summary["last_recorded_at"] is not None
+
+
+def test_summary_of_empty_ledger(ledger: ForwardOutcomeLedger) -> None:
+ assert ledger.summary() == {
+ "total": 0,
+ "by_verdict": {},
+ "by_lineage": {},
+ "evidence": 0,
+ "last_recorded_at": None,
+ }
+
+
+def _event_mission(db: Database, *challengers: str) -> ARCController:
+ """A mission whose BitPro ids are bound by journalled events, not inline projection."""
+ ctrl = ARCController(goal=ARCGoalV1(objective="race outcomes", symbols=["BTC-USDT"]))
+ save_mission(ctrl)
+ for index, challenger in enumerate(challengers):
+ attempt = ARCCandidateAttemptV1(
+ attempt_id=f"attempt-{challenger}",
+ candidate_id=f"candidate-{challenger}",
+ hypothesis=f"variant {challenger}",
+ strategy_code="class Variant: pass\n",
+ strategy_spec={
+ "parameter_changes": {"fast_window": 10 + index},
+ "structural_changes": {},
+ "symbol": "BTC-USDT",
+ "symbols": ["BTC-USDT"],
+ "timeframe": "1H",
+ "parent_strategy_id": 107,
+ },
+ )
+ ctrl.apply_event("candidate_proposed", {"attempt": attempt.model_dump(mode="json")})
+ ctrl.apply_event(
+ "bitpro_self_tested",
+ {
+ "attempt_id": attempt.attempt_id,
+ "bitpro_strategy_id": challenger,
+ "backtest_id": "bt-1",
+ "passed": False,
+ },
+ )
+ with db.session() as session:
+ event_types = set(
+ session.scalars(
+ select(ArcMissionEvent.event_type).where(
+ ArcMissionEvent.mission_id == ctrl.mission_id
+ )
+ ).all()
+ )
+ assert {"candidate_proposed", "bitpro_self_tested"} <= event_types
+ return ctrl
+
+
+def test_lineage_finds_a_mission_that_fell_out_of_the_scan_window(
+ ledger: ForwardOutcomeLedger, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ old = _event_mission(ledger.db, "9301")
+ newer = [_save_mission("9401"), _save_mission("9402"), _save_mission("9403")]
+ with ledger.db.session() as session:
+ stored = session.get(ArcMission, old.mission_id)
+ assert stored is not None
+ stored.updated_at = datetime(2026, 1, 1)
+ for hour, mission in enumerate(newer, start=1):
+ row = session.get(ArcMission, mission.mission_id)
+ assert row is not None
+ row.updated_at = datetime(2026, 2, 1, hour)
+
+ loads: list[str] = []
+ real_load = forward_outcomes.load_projection
+
+ def counting_load(session: Session, mission: ArcMission):
+ loads.append(mission.mission_id)
+ return real_load(session, mission)
+
+ monkeypatch.setattr(forward_outcomes, "load_projection", counting_load)
+
+ narrow = ForwardOutcomeLedger(ledger.db, mission_scan_limit=2)
+ row = _record(narrow, challenger="9301", verdict="won")
+ assert row["lineage_status"] == "linked"
+ assert row["mission_id"] == old.mission_id
+ assert row["changes"] == {
+ "parameter_changes": {"fast_window": 10},
+ "structural_changes": {},
+ }
+ assert loads == [old.mission_id]
+
+
+def test_lineage_prefers_the_newest_mission_for_a_reused_challenger_id(
+ ledger: ForwardOutcomeLedger,
+) -> None:
+ _event_mission(ledger.db, "9302")
+ newest = _event_mission(ledger.db, "9302")
+
+ row = _record(ledger, challenger="9302", verdict="won")
+ assert row["lineage_status"] == "linked"
+ assert row["mission_id"] == newest.mission_id
+ assert row["attempt_id"] == "attempt-9302"
+
+
+def test_event_scan_can_be_disabled_and_falls_back_to_mission_scan(
+ ledger: ForwardOutcomeLedger,
+) -> None:
+ mission = _event_mission(ledger.db, "9303")
+ blind = ForwardOutcomeLedger(ledger.db, event_scan_limit=0)
+
+ row = _record(blind, challenger="9303", verdict="won")
+ assert row["lineage_status"] == "linked"
+ assert row["mission_id"] == mission.mission_id
+
+
+def test_malformed_event_payload_never_breaks_lineage(
+ ledger: ForwardOutcomeLedger,
+) -> None:
+ with ledger.db.session() as session:
+ session.add(
+ ArcMissionEvent(
+ mission_id="arc_malformed",
+ seq=1,
+ event_id="evt_malformed",
+ event_type="bitpro_self_tested",
+ event_json={"payload": "not a dict"},
+ )
+ )
+
+ row = _record(ledger, challenger="9399", verdict="won")
+ assert row["lineage_status"] == "no_lineage"
+
+
+def test_lineage_links_a_numeric_event_challenger_id(
+ ledger: ForwardOutcomeLedger,
+) -> None:
+ # Some payloads carry BitPro ids as JSON numbers. The attempt projection only accepts
+ # strings, so it is persisted inline and the number exists only in the event journal.
+ ctrl = _save_mission("9401")
+ ctrl.apply_event(
+ "bitpro_self_tested",
+ {"bitpro_strategy_id": 9401, "backtest_id": "bt-1", "passed": False},
+ )
+
+ # Disable the fallback mission scan so only the event-index lookup can match.
+ event_only = ForwardOutcomeLedger(ledger.db, mission_scan_limit=0)
+ row = _record(event_only, challenger="9401", verdict="won")
+ assert row["lineage_status"] == "linked"
+ assert row["mission_id"] == ctrl.mission_id
+ assert row["attempt_id"] == "attempt-9401"
diff --git a/tests/test_race_judge.py b/tests/test_race_judge.py
index e066e81..995456e 100644
--- a/tests/test_race_judge.py
+++ b/tests/test_race_judge.py
@@ -3,17 +3,23 @@
from __future__ import annotations
import json
+from datetime import UTC, datetime
from pathlib import Path
+from typing import Any
from unittest.mock import MagicMock, patch
import pytest
+from hypertrade.arc.evolution_models import ArcForwardOutcome
+from hypertrade.arc.forward_outcomes import ForwardOutcomeLedger
from hypertrade.bitpro.mcp import BitProToolAdapter
+from hypertrade.db import Database
from hypertrade.paper.race_judge import (
RaceJudgeDaemon,
RacePairRecord,
build_race_feishu_card,
dispatch_race_feishu_card,
)
+from sqlalchemy import select
def test_race_pair_record_serialization() -> None:
@@ -784,3 +790,235 @@ def test_quantlab_target_without_a_cancellation_field_is_not_an_error(tmp_path:
rec = daemon.evaluate_pair(1, 2, generation=2, target_id="quantlab")
assert rec.target_id == "quantlab" and rec.state == "cancelled"
assert rec.cancellation_kind is None and dispatch.call_args.args[1] == "CANCELLED"
+
+
+# --- Forward outcome ledger: every finished lane is booked exactly once ---
+
+
+def _ledger_daemon(tmp_path: Path, adapter: MagicMock) -> RaceJudgeDaemon:
+ db = Database("sqlite:///:memory:")
+ db.create_all()
+ return RaceJudgeDaemon(
+ db=db,
+ bitpro_adapter=adapter,
+ state_file=tmp_path / "state.json",
+ history_file=tmp_path / "missing-history.json",
+ )
+
+
+def _outcome_rows(daemon: RaceJudgeDaemon) -> list[ArcForwardOutcome]:
+ assert daemon.db is not None
+ with daemon.db.session() as session:
+ return list(session.scalars(select(ArcForwardOutcome)))
+
+
+def _utc(value: datetime | None) -> datetime | None:
+ """SQLite drops tzinfo; the judge always stores UTC."""
+ if value is None or value.tzinfo is not None:
+ return value
+ return value.replace(tzinfo=UTC)
+
+
+_ARCHIVED_EXAM = {
+ "actor": "system:exam",
+ "by_system": True,
+ "kind": "exam",
+ "reason": "终考未过关",
+ "cancelled_at": "2026-10-20T08:00:00+00:00",
+ "detail": {"kind": "exam", "positive_days": 3},
+}
+
+
+def test_futility_elimination_is_recorded_once_in_the_outcome_ledger(tmp_path: Path) -> None:
+ adapter = MagicMock(spec=BitProToolAdapter)
+ adapter.paper_relay_status.return_value = _lane_status(_futility_cancellation())
+ daemon = _ledger_daemon(tmp_path, adapter)
+
+ with patch(
+ "hypertrade.paper.race_judge.dispatch_race_feishu_card", return_value=(True, "ok")
+ ):
+ rec = daemon.evaluate_pair(1, 2, generation=2)
+ assert rec.outcome_recorded is True
+ again = daemon.evaluate_pair(1, 2, generation=2)
+
+ [row] = _outcome_rows(daemon) # a second poll does not add a row
+ assert row.verdict == "lost_futility"
+ assert row.metrics_json["cancellation_metrics"]
+ assert _utc(row.ended_at) == datetime(2026, 10, 20, 8, 0, tzinfo=UTC)
+ assert again.outcome_recorded is True
+
+
+def test_completed_lane_is_recorded_as_a_win_when_the_plan_updates(tmp_path: Path) -> None:
+ adapter = MagicMock(spec=BitProToolAdapter)
+ status = _lane_status(None, state="completed")
+ status["plan"]["updated_at"] = "2026-10-19T12:00:00+00:00"
+ adapter.paper_relay_status.return_value = status
+ daemon = _ledger_daemon(tmp_path, adapter)
+
+ with patch(
+ "hypertrade.paper.race_judge.dispatch_race_feishu_card", return_value=(True, "ok")
+ ):
+ daemon.evaluate_pair(1, 2, generation=2)
+
+ [row] = _outcome_rows(daemon)
+ assert row.verdict == "won" and row.verdict_source == "race_judge"
+ assert _utc(row.ended_at) == datetime(2026, 10, 19, 12, 0, tzinfo=UTC)
+
+
+def test_refilled_seat_with_an_archive_is_mirrored_and_recorded(tmp_path: Path) -> None:
+ adapter = MagicMock(spec=BitProToolAdapter)
+ observing = _lane_status(None, state="observing")
+ observing["proof"] = {"eligible": False, "reason": "观测中", "excess_return_pct": -1.5}
+ adapter.paper_relay_status.return_value = observing
+ daemon = _ledger_daemon(tmp_path, adapter)
+
+ with patch(
+ "hypertrade.paper.race_judge.dispatch_race_feishu_card", return_value=(True, "ok")
+ ) as dispatch:
+ daemon.evaluate_pair(1, 2, generation=2)
+ # BitPro eliminated the lane and refilled its seat before the next poll.
+ adapter.paper_relay_status.return_value = {
+ "plan": None,
+ "proof": {"eligible": True, "excess_return_pct": 9.9, "proof_sha256": "b" * 64},
+ "cancellation": _ARCHIVED_EXAM,
+ }
+ rec = daemon.evaluate_pair(1, 2, generation=2)
+
+ assert rec.state == "cancelled" and rec.cancellation_kind == "exam"
+ dispatch.assert_called_once()
+ assert dispatch.call_args.args[1] == "EXAM_FAILED"
+ [row] = _outcome_rows(daemon)
+ assert row.verdict == "lost_exam"
+ assert _utc(row.ended_at) == datetime(2026, 10, 20, 8, 0, tzinfo=UTC)
+ # The no-plan proof is a fresh candidate evaluation, not this race's evidence.
+ assert row.metrics_json["excess_return_pct"] == -1.5
+ assert row.metrics_json["cancellation_reason"] == "终考未过关"
+
+
+def test_refilled_seat_without_an_archive_ends_neutrally_without_a_card(tmp_path: Path) -> None:
+ adapter = MagicMock(spec=BitProToolAdapter)
+ adapter.paper_relay_status.return_value = _lane_status(None, state="observing")
+ daemon = _ledger_daemon(tmp_path, adapter)
+
+ with patch("hypertrade.paper.race_judge.dispatch_race_feishu_card") as dispatch:
+ daemon.evaluate_pair(1, 2, generation=2)
+ adapter.paper_relay_status.return_value = {
+ "plan": None,
+ "proof": {},
+ "cancellation": None,
+ }
+ rec = daemon.evaluate_pair(1, 2, generation=2)
+
+ dispatch.assert_not_called()
+ assert rec.state == "ended" and rec.outcome_recorded is True
+ [row] = _outcome_rows(daemon)
+ assert row.verdict == "ended_unknown" and row.ended_at is None
+
+
+def test_judge_without_a_database_announces_but_records_nothing(tmp_path: Path) -> None:
+ adapter = MagicMock(spec=BitProToolAdapter)
+ adapter.paper_relay_status.return_value = _lane_status(_futility_cancellation())
+ daemon = _daemon(tmp_path, adapter)
+
+ with patch(
+ "hypertrade.paper.race_judge.dispatch_race_feishu_card", return_value=(True, "ok")
+ ) as dispatch:
+ rec = daemon.evaluate_pair(1, 2, generation=2)
+
+ assert dispatch.call_args.args[1] == "FUTILITY"
+ assert rec.outcome_recorded is False
+ assert daemon.backfill_outcomes() == 0
+
+
+def test_a_failing_ledger_never_blocks_the_card_or_state(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ adapter = MagicMock(spec=BitProToolAdapter)
+ adapter.paper_relay_status.return_value = _lane_status(_futility_cancellation())
+ daemon = _ledger_daemon(tmp_path, adapter)
+
+ def _boom(*args: object, **kwargs: object) -> None:
+ raise RuntimeError("ledger down")
+
+ original = ForwardOutcomeLedger.record
+ monkeypatch.setattr(ForwardOutcomeLedger, "record", _boom)
+ with patch(
+ "hypertrade.paper.race_judge.dispatch_race_feishu_card", return_value=(True, "ok")
+ ) as dispatch:
+ rec = daemon.evaluate_pair(1, 2, generation=2)
+
+ assert dispatch.call_count == 1 and rec.outcome_recorded is False
+ persisted = json.loads((tmp_path / "state.json").read_text(encoding="utf-8"))
+ assert persisted[0]["state"] == "cancelled"
+
+ # The next evaluation retries once the ledger recovers.
+ monkeypatch.setattr(ForwardOutcomeLedger, "record", original)
+ reloaded = RaceJudgeDaemon(
+ db=daemon.db,
+ bitpro_adapter=adapter,
+ state_file=tmp_path / "state.json",
+ history_file=tmp_path / "missing-history.json",
+ )
+ with patch(
+ "hypertrade.paper.race_judge.dispatch_race_feishu_card", return_value=(True, "ok")
+ ):
+ retried = reloaded.evaluate_pair(1, 2, generation=2)
+
+ assert retried.outcome_recorded is True
+ [row] = _outcome_rows(reloaded)
+ assert row.verdict == "lost_futility"
+
+
+def test_legacy_finished_lanes_are_backfilled_once(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ state_file = tmp_path / "state.json"
+ legacy = RacePairRecord(
+ parent_strategy_id=1, challenger_strategy_id=2, generation=2, state="completed"
+ ).to_dict()
+ legacy.pop("outcome_recorded") # a state file written before the ledger existed
+ state_file.write_text(json.dumps([legacy]), encoding="utf-8")
+ db = Database("sqlite:///:memory:")
+ db.create_all()
+ adapter = MagicMock(spec=BitProToolAdapter)
+ adapter.paper_relay_plans.return_value = []
+ daemon = RaceJudgeDaemon(
+ db=db,
+ bitpro_adapter=adapter,
+ state_file=state_file,
+ history_file=tmp_path / "missing-history.json",
+ )
+
+ calls: list[dict[str, Any]] = []
+ original = ForwardOutcomeLedger.record
+
+ def _counting(self: ForwardOutcomeLedger, **kwargs: Any) -> dict[str, Any]:
+ calls.append(kwargs)
+ return original(self, **kwargs)
+
+ monkeypatch.setattr(ForwardOutcomeLedger, "record", _counting)
+ daemon.scan_and_judge_all()
+ daemon.scan_and_judge_all() # backfill runs once per process
+
+ assert len(calls) == 1 and calls[0]["source"] == "backfill"
+ [row] = _outcome_rows(daemon)
+ assert row.verdict == "won" and row.verdict_source == "backfill"
+
+
+def test_a_readmitted_challenger_clears_the_recorded_outcome(tmp_path: Path) -> None:
+ adapter = MagicMock(spec=BitProToolAdapter)
+ adapter.paper_relay_status.return_value = _lane_status(_futility_cancellation())
+ daemon = _ledger_daemon(tmp_path, adapter)
+
+ with patch(
+ "hypertrade.paper.race_judge.dispatch_race_feishu_card", return_value=(True, "ok")
+ ):
+ first = daemon.evaluate_pair(1, 2, generation=2)
+ assert first.outcome_recorded is True
+ # The operator re-admits the same challenger: a new race on the same pair.
+ adapter.paper_relay_status.return_value = _lane_status(None, state="observing")
+ again = daemon.evaluate_pair(1, 2, generation=2)
+
+ assert again.state == "observing" and again.outcome_recorded is False
+ [row] = _outcome_rows(daemon)
+ assert row.verdict == "lost_futility" # the ledger keeps the earlier race's verdict
diff --git a/tests/test_source_variant_avo.py b/tests/test_source_variant_avo.py
index 4153a08..9ff796c 100644
--- a/tests/test_source_variant_avo.py
+++ b/tests/test_source_variant_avo.py
@@ -828,3 +828,48 @@ def test_the_model_is_told_which_changes_were_tried_and_how_many_seats_are_open(
assert "relay.tried_variants" in _SYSTEM and "never\nraces the same changes twice" in _SYSTEM
assert "open (free minus candidates already queued)" in _SYSTEM
+
+
+def test_outcome_memory_is_citable_evidence_and_unknown_outcomes_are_rejected(
+ source_variant_mission,
+):
+ from hypertrade.arc.avo import _SYSTEM, _perform
+
+ context = source_variant_mission.projection.goal.evolution_context
+ context["memory"] = [
+ {
+ "memory_id": "outcome:fo_xxx",
+ "type": "forward_outcome",
+ "evidence_status": "forward_paper_outcome",
+ "verdict": "won",
+ "relation": "made_current",
+ "generations_back": 1,
+ }
+ ]
+ proposed = _perform(
+ source_variant_mission,
+ "propose",
+ {
+ "hypothesis": "stay near the direction that won the seat",
+ "evolution_hypothesis": {**HYPOTHESIS, "evidence_refs": ["outcome:fo_xxx"]},
+ "parameter_changes": {"fast_window": 30},
+ },
+ ARCSelfTestService(),
+ )
+ assert proposed["attempt_id"].startswith("att_avo_")
+
+ with pytest.raises(ValueError, match="cites unavailable evidence"):
+ _perform(
+ source_variant_mission,
+ "propose",
+ {
+ "hypothesis": "cite a race that does not exist",
+ "evolution_hypothesis": {**HYPOTHESIS, "evidence_refs": ["outcome:fo_missing"]},
+ "parameter_changes": {"fast_window": 40},
+ },
+ ARCSelfTestService(),
+ )
+
+ assert "forward_outcome entries" in _SYSTEM
+ assert "made_current" in _SYSTEM
+ assert "noisy observation" in _SYSTEM