From 948f2b002c06d0dde9bea24a6efc74642766aec2 Mon Sep 17 00:00:00 2001 From: shadowell Date: Wed, 21 Oct 2026 11:30:18 +0800 Subject: [PATCH 1/7] =?UTF-8?q?feat(arc):=20=E6=96=B0=E5=A2=9E=E8=B5=9B?= =?UTF-8?q?=E9=A9=AC=E5=89=8D=E5=90=91=E7=BB=93=E6=9E=9C=E8=A1=A8=20arc=5F?= =?UTF-8?q?forward=5Foutcomes=EF=BC=88=E6=A8=A1=E5=9E=8B=20+=200050=20?= =?UTF-8?q?=E8=BF=81=E7=A7=BB=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 每个已结束的 BitPro 赛道一行：母体、挑战者、位号、结论、结束时间、证据摘要与研究血缘 （任务、候选、参数与结构改动、作用范围）。同一 (目标, 母体, 挑战者) 唯一。迁移同时适用于 SQLite 与 PostgreSQL，含 SQLite 迁移测试与隔离 PostgreSQL 上的升降级测试。 Co-Authored-By: Claude Opus 5.5 --- .../versions/0050_arc_forward_outcomes.py | 94 ++++++++++++ .../src/hypertrade/arc/evolution_models.py | 33 ++++- tests/test_arc_forward_outcomes.py | 136 ++++++++++++++++++ tests/test_arc_forward_outcomes_postgres.py | 78 ++++++++++ 4 files changed, 340 insertions(+), 1 deletion(-) create mode 100644 backend/alembic/versions/0050_arc_forward_outcomes.py create mode 100644 tests/test_arc_forward_outcomes.py create mode 100644 tests/test_arc_forward_outcomes_postgres.py 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/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/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..efc56bd --- /dev/null +++ b/tests/test_arc_forward_outcomes_postgres.py @@ -0,0 +1,78 @@ +"""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() From 2bf1f95e7a821ee9a237a981ff479a2568bc9319 Mon Sep 17 00:00:00 2001 From: shadowell Date: Wed, 21 Oct 2026 17:01:00 +0800 Subject: [PATCH 2/7] =?UTF-8?q?feat(arc):=20=E8=B5=9B=E9=A9=AC=E5=89=8D?= =?UTF-8?q?=E5=90=91=E7=BB=93=E6=9E=9C=E8=AE=B0=E8=B4=A6=E4=B8=8E=E7=A0=94?= =?UTF-8?q?=E7=A9=B6=E8=A1=80=E7=BC=98=E5=AF=B9=E6=8E=A5=EF=BC=88ForwardOu?= =?UTF-8?q?tcomeLedger=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 只记已结束赛道：won／lost_futility／lost_exam 是有效输赢证据，超时、被同母体顶替、 操作者取消与来源不明都是中性记录，不当负面证据。 - 同一 (目标, 母体, 挑战者) 只留一行；具体结论写入后不再改写，只有「来源不明」可被更具体的 结论升级；并发插入撞唯一约束时回滚并按已有行对齐。 - 血缘按 BitPro 策略 id 对上 ARC 尝试（最新任务优先，找全即停），一次扫描可批量对接； record(link=False) 供补录先记账、再统一对接；对不上的标 no_lineage，不推断。 - summary 用 SQL 聚合，不把整表读进内存。 由 DeepSeek V4.1 Flash 按 Claude 的任务说明实现，经 Claude 审查一轮后修改。 Co-Authored-By: Claude Opus 5.5 --- .../src/hypertrade/arc/forward_outcomes.py | 385 +++++++++++++ tests/test_forward_outcomes_ledger.py | 510 ++++++++++++++++++ 2 files changed, 895 insertions(+) create mode 100644 backend/src/hypertrade/arc/forward_outcomes.py create mode 100644 tests/test_forward_outcomes_ledger.py diff --git a/backend/src/hypertrade/arc/forward_outcomes.py b/backend/src/hypertrade/arc/forward_outcomes.py new file mode 100644 index 0000000..53e7672 --- /dev/null +++ b/backend/src/hypertrade/arc/forward_outcomes.py @@ -0,0 +1,385 @@ +"""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, 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") + +_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"), + } + + +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) -> None: + self.db = db + self.mission_scan_limit = mission_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 _lineage_index( + self, + session: Session, + wanted: set[str], + ) -> dict[str, tuple[ArcMission, ARCCandidateAttemptV1]]: + """Newest mission per wanted challenger; stop loading projections once all are found.""" + found: dict[str, tuple[ArcMission, ARCCandidateAttemptV1]] = {} + if not wanted: + return found + missions = session.scalars( + select(ArcMission) + .order_by(ArcMission.updated_at.desc()) + .limit(self.mission_scan_limit) + ).yield_per(20) + for mission in missions: + projection = load_projection(session, mission) + for attempt in projection.attempts: + challenger = str(attempt.bitpro_strategy_id) + if challenger in wanted 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_parent( + self, + *, + target_id: str, + parent_strategy_id: object, + limit: int = 10, + ) -> list[dict[str, Any]]: + """Concrete, lineage-linked outcomes for one parent; neutral records never qualify.""" + rows = select(ArcForwardOutcome).where( + ArcForwardOutcome.target_id == target_id, + ArcForwardOutcome.parent_strategy_id == str(parent_strategy_id), + ArcForwardOutcome.verdict.in_(EVIDENCE_VERDICTS), + ArcForwardOutcome.lineage_status == "linked", + ) + statement = rows.order_by( + case((ArcForwardOutcome.ended_at.is_(None), 1), else_=0), + ArcForwardOutcome.ended_at.desc(), + ArcForwardOutcome.created_at.desc(), + ).limit(max(0, limit)) + with self.db.session() as session: + return [self.view(row) for row in session.scalars(statement).all()] + + 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/tests/test_forward_outcomes_ledger.py b/tests/test_forward_outcomes_ledger.py new file mode 100644 index 0000000..e6da75f --- /dev/null +++ b/tests/test_forward_outcomes_ledger.py @@ -0,0 +1,510 @@ +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, 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_parent_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_parent(target_id="bitpro", parent_strategy_id=107) + 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) + + limited = ledger.evidence_for_parent( + target_id="bitpro", parent_strategy_id=107, limit=2 + ) + assert [row["challenger_strategy_id"] for row in limited] == ["9002", "9001"] + + +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, + } From 3f0c2b1e647a225b61b82b85e10ed0f13f945654 Mon Sep 17 00:00:00 2001 From: shadowell Date: Wed, 21 Oct 2026 17:12:41 +0800 Subject: [PATCH 3/7] =?UTF-8?q?feat(race-judge):=20=E8=B5=9B=E9=81=93?= =?UTF-8?q?=E7=BB=93=E6=9D=9F=E6=97=B6=E8=AE=B0=E5=85=A5=E5=89=8D=E5=90=91?= =?UTF-8?q?=E7=BB=93=E6=9E=9C=E8=B4=A6=E6=9C=AC=EF=BC=9B=E4=BD=8D=E7=BD=AE?= =?UTF-8?q?=E8=A2=AB=E8=A1=A5=E8=B5=B0=E6=97=B6=E6=8C=89=E6=B7=98=E6=B1=B0?= =?UTF-8?q?=E5=BD=92=E6=A1=A3=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 赛道变为 completed／cancelled／ended 时写入 arc_forward_outcomes（completed 用计划的 updated_at、cancelled 用归档的 cancelled_at 作为结束时间）；记账失败只记日志、下次再试， 不影响卡片、状态文件与轮询。 - 计划消失但 relay-control 带回淘汰归档时（BitPro 已把位置补给下一个候选、删了旧计划行）， 按取消处理：照常发终考未过／超时等卡片并记输赢，数字保留上次镜像的、不用候选重评的； 不带归档（旧 BitPro）时仍记为已结束。 - 重新放行的挑战者复位记账标记（账本不改写第一场的具体结论，已知限制）；进程首次巡检前 补录状态文件里已结束的赛道，一次扫描批量对接研究血缘。 - 抽出 _mirror_cancellation 与 _announce，行为不变。 由 DeepSeek V4.1 Flash 按 Claude 的任务说明实现，经 Claude 审查与变异检验（4 处守卫均被用例抓住）。 Co-Authored-By: Claude Opus 5.5 --- backend/src/hypertrade/paper/race_judge.py | 157 +++++++++++++- tests/test_race_judge.py | 238 +++++++++++++++++++++ 2 files changed, 388 insertions(+), 7 deletions(-) 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/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 From 21aadf1be5fd67db24546cf3ac12d45c58e48a2e Mon Sep 17 00:00:00 2001 From: shadowell Date: Thu, 22 Oct 2026 00:10:23 +0800 Subject: [PATCH 4/7] =?UTF-8?q?feat(arc):=20=E8=B5=9B=E9=A9=AC=E5=89=8D?= =?UTF-8?q?=E5=90=91=E7=BB=93=E6=9E=9C=E6=B2=BF=E6=8E=A5=E7=8F=AD=E9=93=BE?= =?UTF-8?q?=E5=9B=9E=E6=B5=81=E7=A0=94=E7=A9=B6=E8=AE=B0=E5=BF=86=EF=BC=8C?= =?UTF-8?q?=E6=8F=90=E7=A4=BA=E6=A8=A1=E5=9E=8B=E6=80=8E=E4=B9=88=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ForwardOutcomeLedger.evidence_for_lineage 取代 evidence_for_parent：赢家接班后编号会变， 它赢下的那场和前任们的比赛都记在前任名下，所以沿 won 记录往回最多 3 代，取已对接血缘的 有效证据（最多 8 条），每条标 relation（made_current／variant_of_current／made_ancestor／ variant_of_ancestor）与 generations_back；防环。 - EvolutionService._memory 把这些结果作为 type=forward_outcome 的记忆条目放在开发记忆之前 （模型只看前 20 条），manifest 记 forward_outcomes 计数；账本出错记运行错误、研究照常。 - AVO 系统提示说明各 relation 的含义、改动相对那场比赛的母体、输家方向无新证据不重复、 赢的方向附近优先、单场是噪声观察不是因果结论；outcome:… 可在 evidence_refs 引用。 - 研究证据视图：memory_references 只列开发参照，前向结果单列 forward_outcome_references。 由 DeepSeek V4.1 Flash 按 Claude 的任务说明实现，经两轮审查与变异检验（4 处均被用例抓住）。 Co-Authored-By: Claude Opus 5.5 --- backend/src/hypertrade/arc/avo.py | 12 ++ backend/src/hypertrade/arc/evidence_view.py | 19 ++ backend/src/hypertrade/arc/evolution.py | 77 +++++++- .../src/hypertrade/arc/forward_outcomes.py | 77 ++++++-- tests/test_arc_evolution.py | 175 ++++++++++++++++++ tests/test_avo_evidence.py | 46 +++++ tests/test_forward_outcomes_ledger.py | 69 ++++++- tests/test_source_variant_avo.py | 45 +++++ 8 files changed, 502 insertions(+), 18 deletions(-) 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..f2c3b87 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 ( @@ -1100,4 +1101,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/forward_outcomes.py b/backend/src/hypertrade/arc/forward_outcomes.py index 53e7672..60205b3 100644 --- a/backend/src/hypertrade/arc/forward_outcomes.py +++ b/backend/src/hypertrade/arc/forward_outcomes.py @@ -312,27 +312,78 @@ def link_pending(self, limit: int = 20) -> int: session.flush() return len(rows) - def evidence_for_parent( + def evidence_for_lineage( self, *, target_id: str, - parent_strategy_id: object, - limit: int = 10, + strategy_id: object, + ancestors: int = 3, + limit: int = 8, ) -> list[dict[str, Any]]: - """Concrete, lineage-linked outcomes for one parent; neutral records never qualify.""" - rows = select(ArcForwardOutcome).where( - ArcForwardOutcome.target_id == target_id, - ArcForwardOutcome.parent_strategy_id == str(parent_strategy_id), - ArcForwardOutcome.verdict.in_(EVIDENCE_VERDICTS), - ArcForwardOutcome.lineage_status == "linked", - ) - statement = rows.order_by( + """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(), - ).limit(max(0, limit)) + ) + chain = [str(strategy_id)] with self.db.session() as session: - return [self.view(row) for row in session.scalars(statement).all()] + 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.""" diff --git a/tests/test_arc_evolution.py b/tests/test_arc_evolution.py index c291cf9..0750b7e 100644 --- a/tests/test_arc_evolution.py +++ b/tests/test_arc_evolution.py @@ -806,3 +806,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_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 index e6da75f..82bed48 100644 --- a/tests/test_forward_outcomes_ledger.py +++ b/tests/test_forward_outcomes_ledger.py @@ -433,7 +433,9 @@ def get_once_none(self, entity, ident, **kwargs): assert rows[0].verdict == "lost_exam" -def test_evidence_for_parent_filters_sorts_and_limits(ledger: ForwardOutcomeLedger) -> None: +def test_evidence_for_lineage_ancestors_zero_filters_sorts_and_limits( + ledger: ForwardOutcomeLedger, +) -> None: _save_mission("9001", "9002", "9003", "9004", "9006") _record( ledger, @@ -469,17 +471,76 @@ def test_evidence_for_parent_filters_sorts_and_limits(ledger: ForwardOutcomeLedg ) _record(ledger, challenger="9006", verdict="won", ended_at=None) - results = ledger.evidence_for_parent(target_id="bitpro", parent_strategy_id=107) + 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_parent( - target_id="bitpro", parent_strategy_id=107, limit=2 + 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") 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 From 8283c5f8e4420ecf58309defc98d7f16c6f298de Mon Sep 17 00:00:00 2001 From: shadowell Date: Thu, 22 Oct 2026 09:02:06 +0800 Subject: [PATCH 5/7] =?UTF-8?q?fix(arc):=20=E7=A0=94=E7=A9=B6=E8=A1=80?= =?UTF-8?q?=E7=BC=98=E5=85=88=E6=8C=89=E4=BA=8B=E4=BB=B6=E7=B4=A2=E5=BC=95?= =?UTF-8?q?=E5=AE=9A=E4=BD=8D=EF=BC=8C=E7=A0=94=E7=A9=B6=E6=8F=90=E9=80=9F?= =?UTF-8?q?=E5=90=8E=E6=97=A7=E4=BB=BB=E5=8A=A1=E4=B9=9F=E8=83=BD=E5=AF=B9?= =?UTF-8?q?=E4=B8=8A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 研究预算提到每天最多 32 轮后，「按最近更新的 500 个任务逐个加载投影」只覆盖约半个月， 而变体可能排队几天再赛 14～17 天，结束时当初的任务已滑出窗口，被永久记成 no_lineage。 - 新增 _locate_by_events：只在 candidate_proposed／bitpro_self_tested／candidate_validated 三类事件里，用 JSON 路径取编号字段（不读含整份策略代码的事件正文），按事件 id 倒序流式 读取（PostgreSQL 走服务端游标），最新事件优先，全部找到即停并关闭游标；数字编号统一 按字符串比较。 - _lineage_index：先按事件定位、每个命中任务只加载一次投影；仍未找到的再回退扫描最近 任务（覆盖事件内联存储的早期任务），已加载过的跳过。 - 测试：旧任务滑出扫描窗口仍能对上且只加载一次投影、重复使用的编号取最新任务、只能靠 回退的任务、数字编号、载荷异常不报错；PostgreSQL 真库验证 JSON 路径查询与排除无关事件。 由 DeepSeek V4.1 Flash 按 Claude 的任务说明实现，经两轮审查与变异检验（4 处均被用例抓住）。 Co-Authored-By: Claude Opus 5.5 --- .../src/hypertrade/arc/forward_outcomes.py | 96 +++++++++++- tests/test_arc_forward_outcomes_postgres.py | 45 ++++++ tests/test_forward_outcomes_ledger.py | 137 +++++++++++++++++- 3 files changed, 273 insertions(+), 5 deletions(-) diff --git a/backend/src/hypertrade/arc/forward_outcomes.py b/backend/src/hypertrade/arc/forward_outcomes.py index 60205b3..614f3b1 100644 --- a/backend/src/hypertrade/arc/forward_outcomes.py +++ b/backend/src/hypertrade/arc/forward_outcomes.py @@ -24,7 +24,7 @@ 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, Database +from hypertrade.db import ArcMission, ArcMissionEvent, Database logger = logging.getLogger("hypertrade.arc.forward_outcomes") @@ -40,6 +40,10 @@ # 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", @@ -154,12 +158,25 @@ def _apply_lineage( } +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) -> None: + 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, @@ -253,25 +270,96 @@ def _reconcile( 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; stop loading projections once all are found.""" + """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 wanted and challenger not in found: + if challenger in missing and challenger not in found: found[challenger] = (mission, attempt) if len(found) == len(wanted): return found diff --git a/tests/test_arc_forward_outcomes_postgres.py b/tests/test_arc_forward_outcomes_postgres.py index efc56bd..9058ae8 100644 --- a/tests/test_arc_forward_outcomes_postgres.py +++ b/tests/test_arc_forward_outcomes_postgres.py @@ -76,3 +76,48 @@ def test_postgres_upgrade_write_unique_pair_and_downgrade(tmp_path) -> None: 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_forward_outcomes_ledger.py b/tests/test_forward_outcomes_ledger.py index 82bed48..0b4ee40 100644 --- a/tests/test_forward_outcomes_ledger.py +++ b/tests/test_forward_outcomes_ledger.py @@ -16,7 +16,7 @@ verdict_for, ) from hypertrade.arc.store import configure_store, reset_store, save_mission -from hypertrade.db import ArcMission, Database +from hypertrade.db import ArcMission, ArcMissionEvent, Database from hypertrade.paper.race_judge import RacePairRecord from sqlalchemy import select from sqlalchemy.orm import Session @@ -569,3 +569,138 @@ def test_summary_of_empty_ledger(ledger: ForwardOutcomeLedger) -> None: "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" From 146444c104f233c6fa499b68931fe23aae0f40c3 Mon Sep 17 00:00:00 2001 From: shadowell Date: Thu, 22 Oct 2026 09:04:52 +0800 Subject: [PATCH 6/7] =?UTF-8?q?feat(arc):=20=E8=BF=9B=E5=8C=96=E7=8A=B6?= =?UTF-8?q?=E6=80=81=E6=8E=A5=E5=8F=A3=E9=99=84=E4=B8=8A=E5=89=8D=E5=90=91?= =?UTF-8?q?=E7=BB=93=E6=9E=9C=E7=BB=9F=E8=AE=A1=EF=BC=8C=E6=96=87=E6=A1=A3?= =?UTF-8?q?=E5=90=8C=E6=AD=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - GET /api/v1/arc/evolution 在接口层附上 forward_outcomes（总数、按结论、按血缘、有效证据数、 最近记账时间；读取失败只带 error、接口照常 200）。不改 EvolutionService.status()：研究扫描 每轮内部调用它多次，用不上这份统计。 - 合同文档：「消失的赛道」改为带淘汰归档时按淘汰处理；新增「赛马前向结果回流研究记忆 （2026-10-21）」一节。架构文档、三个 README 的迁移数量 49→50；spec 与 progress 同步。 由 DeepSeek V4.1 Flash 按 Claude 的任务说明实现，经 Claude 审查。 Co-Authored-By: Claude Opus 5.5 --- README.en.md | 4 +- README.md | 4 +- README.zh-CN.md | 4 +- backend/src/hypertrade/arc/evolution.py | 11 +++++ .../src/hypertrade/arc/evolution_router.py | 9 ++++- docs/architecture/33-system-architecture.md | 8 ++-- .../user-directed-evolution-hardening.md | 21 +++++++++- docs/progress.md | 7 ++++ docs/spec.md | 2 +- tests/test_arc_evolution.py | 2 + tests/test_arc_external_surface.py | 40 +++++++++++++++++++ 11 files changed, 98 insertions(+), 14 deletions(-) diff --git a/README.en.md b/README.en.md index 9ec0fb9..f1f790e 100644 --- a/README.en.md +++ b/README.en.md @@ -56,7 +56,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"] @@ -188,7 +188,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 5dfa411..4907e19 100644 --- a/README.md +++ b/README.md @@ -72,7 +72,7 @@ flowchart LR ```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["自主进化引擎（默认开启）
退化扫描 → 预算准入 → 再研究 → 赛马对决 → 效果账本"] @@ -204,7 +204,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/README.zh-CN.md b/README.zh-CN.md index 51f3d99..9bd83fb 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -56,7 +56,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["自主进化引擎（默认开启）
退化扫描 → 预算准入 → 再研究 → 赛马对决 → 效果账本"] @@ -188,7 +188,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/src/hypertrade/arc/evolution.py b/backend/src/hypertrade/arc/evolution.py index f2c3b87..7b47b8d 100644 --- a/backend/src/hypertrade/arc/evolution.py +++ b/backend/src/hypertrade/arc/evolution.py @@ -273,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 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/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 2dd008b..2bba9f0 100644 --- a/docs/progress.md +++ b/docs/progress.md @@ -1,3 +1,10 @@ +## 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`。 +- **验证**：`LC_ALL=en_US.UTF-8 uv run pytest tests/test_arc_evolution.py tests/test_forward_outcomes_ledger.py tests/test_evolution_budget.py tests/test_arc_external_surface.py -q` → 147 passed；`uv run ruff check backend tests` → All checks passed；`uv run mypy backend/src` → Success: no issues found in 317 source files。未运行 `./scripts/check.sh`（由审查方执行）。 +- **未做**：输家逐项消融为第二期；未线上实测。 + ## 2026-10-20：研究侧跟上 BitPro 变体位（按空位扫描、参数去重、需求可见） - **背景**：BitPro #1286 让每个母体默认有三个变体位，研究侧还是按「一个母体一个挑战者」工作：空位已被排队候选预订时仍会继续造候选；同一改动换一天就绕过「已归档实验」检查（开发窗口随日期移动），BitPro 又不会让完全相同的变体赛两次，副本白白花掉模型调用与回测；也看不到空位需求与每日预算的差距。 diff --git a/docs/spec.md b/docs/spec.md index 6e91ec1..1da6ac4 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 0750b7e..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") 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"] From d493de47de92e133d7fca62cd386eccc1e5ddaa1 Mon Sep 17 00:00:00 2001 From: shadowell Date: Thu, 22 Oct 2026 09:11:53 +0800 Subject: [PATCH 7/7] =?UTF-8?q?docs(progress):=20=E8=AE=B0=E5=BD=95?= =?UTF-8?q?=E6=95=B4=E5=A5=97=E6=A3=80=E6=9F=A5=E4=B8=8E=E5=8F=98=E5=BC=82?= =?UTF-8?q?=E6=A3=80=E9=AA=8C=E7=BB=93=E6=9E=9C=E3=80=81=E7=A0=94=E7=A9=B6?= =?UTF-8?q?=E4=B8=BB=E5=BE=AA=E7=8E=AF=E7=9A=84=E5=BD=B1=E5=93=8D=E5=88=86?= =?UTF-8?q?=E6=9E=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- docs/progress.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/progress.md b/docs/progress.md index 8fbc90e..2b17158 100644 --- a/docs/progress.md +++ b/docs/progress.md @@ -2,7 +2,7 @@ - **背景**：赛马结束的结论此前只留在裁判状态文件里，研究侧看不到自己提的改动实际赛成什么样；研究提速到每天最多 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`。 -- **验证**：`LC_ALL=en_US.UTF-8 uv run pytest tests/test_arc_evolution.py tests/test_forward_outcomes_ledger.py tests/test_evolution_budget.py tests/test_arc_external_surface.py -q` → 147 passed；`uv run ruff check backend tests` → All checks passed；`uv run mypy backend/src` → Success: no issues found in 317 source files。未运行 `./scripts/check.sh`（由审查方执行）。 +- **验证**：新增 `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）