feat(db): record the stop a tranche was sized against (#520, #502 stage 2)

keel/agent.py

@@ -320,9 +320,16 @@ def _open_tranche(
     order: dict[str, Any] | None,
     result: ExecutionResult,
     now_ts: int,
+    initial_stop: Decimal | None = None,
 ) -> None:
     """Record the newly opened tranche in the `positions` ledger and point it at its bracket.
 
+    `initial_stop` is the stop this tranche was SIZED against (#520). It is recorded at OPEN and
+    never rewritten, because that is the whole point: `open_stop:<product_id>` already tracks the
+    current, ratcheting stop, and the break-even threshold needs the ORIGINAL per-unit risk.
+    `None` is legitimate and means unknown -- DCA carries no stop by design -- and readers must
+    disable the break-even arm for such a tranche rather than substitute the current stop.
+
     The ledger is what a later exit attributes P&L against, so a tranche whose entry price or qty
     could not be read is NOT recorded: `record_closed_trade` already refuses to guess a missing
     entry price, and a ledger row carrying `None` would either crash the arithmetic or fabricate
@@ -366,6 +373,7 @@ def _open_tranche(
         # Nothing else preserves it: the exit's order row knows only its own fee.
         entry_fee=order["fee"] or Decimal("0"),
         entry_fill=entry_fill,
+        initial_stop=initial_stop,
     )
     if result.bracket_order_id is not None:
         repo.set_position_bracket(position_id, result.bracket_order_id)
@@ -1640,7 +1648,19 @@ def run_once(
                         f"position_rule:{product_id}",
                         {"rule_name": signal.rule_name, "opened_at": now_ts},
                     )
-                    _open_tranche(repo, product_id, signal.rule_name, order, result, now_ts)
+                    _open_tranche(
+                        repo,
+                        product_id,
+                        signal.rule_name,
+                        order,
+                        result,
+                        now_ts,
+                        # #520: the stop the position was SIZED against, captured at open
+                        # because nothing else preserves it -- `open_stop:<product>` starts
+                        # here and then ratchets away from it. `None` for DCA, which has no
+                        # stop by design.
+                        initial_stop=signal.setup.stop if signal.setup is not None else None,
+                    )
 
         cycle_result = LoopResult(
             ts=now_ts,

keel/data/db.py

@@ -19,7 +19,7 @@
 from pathlib import Path
 from typing import Any
 
-SCHEMA_VERSION = 11
+SCHEMA_VERSION = 12
 
 # Creation order matters for readability (and for backends that validate FK targets eagerly);
 # SQLite itself only checks FK targets at DML time, but we still declare referenced tables first.
@@ -84,6 +84,7 @@
         qty               TEXT    NOT NULL,
         entry_fill        TEXT    NOT NULL,
         entry_fee         TEXT    NOT NULL,
+        initial_stop      TEXT,
         bracket_order_id  INTEGER,
         status            TEXT    NOT NULL DEFAULT 'open',
         FOREIGN KEY (bracket_order_id) REFERENCES orders(id)
@@ -472,6 +473,35 @@ def _migrate_v11_orders_filled_quantity(conn: sqlite3.Connection) -> None:
         conn.execute("ALTER TABLE orders ADD COLUMN filled_quantity TEXT")
 
 
+def _migrate_v12_positions_initial_stop(conn: sqlite3.Connection) -> None:
+    """v12 adds `positions.initial_stop` -- the stop the tranche was SIZED against (#520).
+
+    The break-even arm of `exit_policy.next_stop` computes its threshold from the trade's
+    ORIGINAL per-unit risk: `entry + be_roll_rr * (entry - initial_stop)`. Live state carries
+    `entry_fill` (this ledger) and `open_stop:<product_id>` (the CURRENT, already-ratcheted stop)
+    and nothing else -- so the number the threshold is most sensitive to was simply absent.
+
+    Substituting the current stop is not an approximation, it is a DIFFERENT POLICY: the current
+    stop rises on every ratchet, shrinking `(entry - stop)` so the threshold creeps toward entry
+    and the arm fires earlier each time, drifting further from the measured policy the longer a
+    trade runs. Live and sim would then encode two different break-even rules while appearing to
+    share `exit_policy`'s functions -- the exact failure sharing them was meant to prevent.
+
+    Idempotent by the v8/v11 `PRAGMA table_info` guard: a database already stamped at v11 got
+    `positions` from v4's DDL, which has no such column, and `CREATE TABLE IF NOT EXISTS` never
+    adds one.
+
+    **NO BACKFILL, deliberately.** The honest value for every pre-v12 tranche is NULL -- nobody
+    recorded it — and readers must treat NULL as "unknown" and disable the break-even arm for
+    that tranche rather than guess. Inventing a value here would fabricate the one input the
+    policy is most sensitive to. The trailing arm is unaffected: it needs no original risk, so an
+    old tranche keeps trailing and simply never break-even-rolls.
+    """
+    columns = {row["name"] for row in conn.execute("PRAGMA table_info(positions)")}
+    if "initial_stop" not in columns:
+        conn.execute("ALTER TABLE positions ADD COLUMN initial_stop TEXT")
+
+
 _MIGRATIONS: dict[int, Callable[[sqlite3.Connection], None]] = {
     2: _migrate_v2_broker_subscriptions,
     3: _migrate_v3_trade_outcomes,
@@ -483,6 +513,7 @@ def _migrate_v11_orders_filled_quantity(conn: sqlite3.Connection) -> None:
     9: _migrate_v9_screen_exceptions,
     10: _migrate_v10_instrument_attestations,
     11: _migrate_v11_orders_filled_quantity,
+    12: _migrate_v12_positions_initial_stop,
 }
 
 

keel/data/repository.py

@@ -455,9 +455,7 @@ def upsert_broker_subscription(self, record: BrokerSubscription) -> None:
     def list_broker_subscriptions(self) -> list[BrokerSubscription]:
         """Every attested subscription, ordered by venue -- what `keel subscription show`
         renders."""
-        rows = self._conn.execute(
-            "SELECT * FROM broker_subscriptions ORDER BY venue"
-        ).fetchall()
+        rows = self._conn.execute("SELECT * FROM broker_subscriptions ORDER BY venue").fetchall()
         return [_subscription_from_row(row) for row in rows]
 
     # -- trade outcomes (closed round-trips; rails 11 and 16) ---------------
@@ -529,15 +527,24 @@ def open_position(
         qty: Decimal,
         entry_fill: Decimal,
         entry_fee: Decimal,
+        initial_stop: Decimal | None = None,
         bracket_order_id: int | None = None,
     ) -> int:
-        """Record a newly opened tranche and return its id."""
+        """Record a newly opened tranche and return its id.
+
+        `initial_stop` is the stop this tranche was SIZED against (#520) -- the original
+        per-unit risk the break-even threshold is computed from. `None` is a legitimate value
+        and means "unknown", not "zero": DCA has no stop by design, and every tranche opened
+        before v12 predates the column. Readers must disable the break-even arm for such a
+        tranche rather than substitute the current stop, which is a different policy (see
+        `db._migrate_v12_positions_initial_stop`).
+        """
         cursor = self._conn.execute(
             """
             INSERT INTO positions
                 (product_id, rule_name, opened_at, qty, entry_fill, entry_fee,
-                 bracket_order_id, status)
-            VALUES (?, ?, ?, ?, ?, ?, ?, 'open')
+                 initial_stop, bracket_order_id, status)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'open')
             """,
             (
                 product_id,
@@ -546,6 +553,7 @@ def open_position(
                 _dec_to_text(qty),
                 _dec_to_text(entry_fill),
                 _dec_to_text(entry_fee),
+                None if initial_stop is None else _dec_to_text(initial_stop),
                 bracket_order_id,
             ),
         )
@@ -607,6 +615,12 @@ def _position_row_to_dict(self, row: sqlite3.Row) -> dict[str, Any]:
         d = dict(row)
         for field in ("qty", "entry_fill", "entry_fee"):
             d[field] = _text_to_dec(d[field])
+        # `initial_stop` (#520) decodes the same way but is NULLABLE, and the distinction is
+        # load-bearing: `None` means "nobody recorded it" -- DCA, or a tranche predating v12 --
+        # and the break-even arm must switch OFF for it rather than substitute a value.
+        # `_text_to_dec` is not asked to invent a zero.
+        raw_initial_stop = d.get("initial_stop")
+        d["initial_stop"] = None if raw_initial_stop is None else _text_to_dec(raw_initial_stop)
         return d
 
     # -- profile (the user's own settings) ------------------------------------

tests/data/test_db.py

@@ -103,11 +103,11 @@ def test_agent_state_table_has_key_primary_key():
     assert pk_columns == {"key"}
 
 
-def test_schema_version_is_11():
+def test_schema_version_is_12():
     """Deliberate tripwire: bump this literal consciously on every schema change."""
     from keel.data.db import SCHEMA_VERSION
 
-    assert SCHEMA_VERSION == 11
+    assert SCHEMA_VERSION == 12
 
 
 def test_a_v6_database_migrates_up_and_gains_the_profile_table(tmp_path):

tests/data/test_initial_stop.py

@@ -0,0 +1,105 @@
+"""`positions.initial_stop` — the stop a tranche was SIZED against (#520).
+
+The break-even arm of `exit_policy.next_stop` needs the trade's ORIGINAL per-unit risk
+(`entry + be_roll_rr * (entry - initial_stop)`). Live state carried `entry_fill` and the current,
+ratcheting `open_stop:<product_id>` and nothing else, so the number the threshold is most
+sensitive to was simply absent.
+
+The distinction these tests defend: `None` means UNKNOWN, and a reader must disable the
+break-even arm for that tranche rather than substitute the current stop. Substituting is not an
+approximation — the current stop rises on every ratchet, so the threshold would creep toward
+entry and the arm would fire earlier each time, drifting further from the measured policy the
+longer a trade runs.
+"""
+
+from __future__ import annotations
+
+from decimal import Decimal
+
+from keel.data import db
+from keel.data.repository import Repository
+
+
+def _repo() -> Repository:
+    conn = db.connect(":memory:")
+    db.migrate(conn)
+    return Repository(conn)
+
+
+def _open(repo: Repository, **overrides):  # noqa: ANN003, ANN202
+    kwargs = dict(
+        product_id="BTC-USD",
+        rule_name="pullback_continuation",
+        opened_at=1_787_000_000,
+        qty=Decimal("0.01"),
+        entry_fill=Decimal("50000"),
+        entry_fee=Decimal("0.5"),
+    )
+    kwargs.update(overrides)
+    return repo.open_position(**kwargs)
+
+
+def test_a_recorded_initial_stop_round_trips_as_a_decimal() -> None:
+    repo = _repo()
+    _open(repo, initial_stop=Decimal("49000"))
+
+    position = repo.get_open_positions("BTC-USD")[0]
+    assert position["initial_stop"] == Decimal("49000")
+    assert isinstance(position["initial_stop"], Decimal)
+
+
+def test_an_unrecorded_initial_stop_reads_as_None_not_zero() -> None:
+    """The load-bearing distinction. Zero would be a stop 100% below entry -- a real number, and
+    a catastrophic one to compute a break-even threshold from. `None` says 'nobody recorded it'."""
+    repo = _repo()
+    _open(repo)  # no initial_stop -- DCA, or any caller that has none
+
+    position = repo.get_open_positions("BTC-USD")[0]
+    assert position["initial_stop"] is None
+
+
+def test_dca_style_open_without_a_stop_is_accepted_not_rejected() -> None:
+    """DCA has no stop BY DESIGN. Requiring one would refuse a legitimate tranche."""
+    repo = _repo()
+    position_id = _open(repo, rule_name="dca", initial_stop=None)
+
+    assert position_id is not None
+    assert repo.get_open_positions("BTC-USD")[0]["initial_stop"] is None
+
+
+def test_the_column_exists_on_a_freshly_migrated_database() -> None:
+    conn = db.connect(":memory:")
+    db.migrate(conn)
+    columns = {row["name"] for row in conn.execute("PRAGMA table_info(positions)")}
+    assert "initial_stop" in columns
+
+
+def test_the_migration_is_idempotent_and_does_not_backfill() -> None:
+    """A database stamped before v12 got `positions` from v4's DDL, which has no such column, and
+    `CREATE TABLE IF NOT EXISTS` never adds one -- hence the `PRAGMA table_info` guard.
+
+    The honest value for a pre-v12 tranche is NULL. A backfill would fabricate the one input the
+    policy is most sensitive to.
+    """
+    conn = db.connect(":memory:")
+    db.migrate(conn)
+    repo = Repository(conn)
+    _open(repo, initial_stop=Decimal("49000"))
+
+    db.migrate(conn)  # running it again must not raise, duplicate the column, or rewrite data
+
+    columns = [row["name"] for row in conn.execute("PRAGMA table_info(positions)")]
+    assert columns.count("initial_stop") == 1
+    assert repo.get_open_positions("BTC-USD")[0]["initial_stop"] == Decimal("49000")
+
+
+def test_initial_stop_is_not_rewritten_when_the_bracket_moves() -> None:
+    """`open_stop:<product>` ratchets; this does not. That separation IS the fix -- if the ledger
+    tracked the current stop there would be nothing to compute the original risk from."""
+    repo = _repo()
+    position_id = _open(repo, initial_stop=Decimal("49000"))
+
+    repo.set_position_bracket(position_id, None)
+    repo.set_state("open_stop:BTC-USD", Decimal("50500"))  # a ratchet moved the live stop
+
+    assert repo.get_open_positions("BTC-USD")[0]["initial_stop"] == Decimal("49000")

tests/data/test_migrations.py

@@ -46,7 +46,7 @@ def test_fresh_database_is_stamped_at_the_current_version() -> None:
     conn = db.connect(":memory:")
     db.migrate(conn)
     version = conn.execute("SELECT version FROM schema_version").fetchone()["version"]
-    assert version == db.SCHEMA_VERSION == 11
+    assert version == db.SCHEMA_VERSION == 12
 
 
 def test_fresh_database_gets_no_subscription_row() -> None:

tests/data/test_trade_outcomes.py

@@ -35,11 +35,11 @@ def _outcome(**overrides: object) -> dict:
     return base
 
 
-def test_schema_is_at_version_11() -> None:
+def test_schema_is_at_version_12() -> None:
     conn = db.connect(":memory:")
     db.migrate(conn)
     version = conn.execute("SELECT version FROM schema_version").fetchone()["version"]
-    assert version == db.SCHEMA_VERSION == 11
+    assert version == db.SCHEMA_VERSION == 12
 
 
 def test_fresh_database_has_no_outcomes() -> None: