Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions blueprints/us-equities/adaptive-paper/native_adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -714,6 +714,10 @@ async def reconcile_fills(self, cid, target=None):
average = prior["averages"].get(cum, (notional / cum).quantize(Decimal("0.000001")))
result = sink_observation({"client_order_id": cid, "id": prior["id"], "status": status,
"filled_qty": str(cum), "filled_avg_price": str(average), "updated_at_ns": stamp,
# https://docs.alpaca.markets/docs/account-activities:
# FILL transaction_time is the
# execution time. Preserve its integer ns through the sink.
"execution_time_ns": stamp,
"event": "fill" if status == "filled" else "partial_fill",
"execution_id": execution["trade_id"], "event_qty": str(execution["qty"]),
"event_price": str(execution["price"])})
Expand Down Expand Up @@ -892,6 +896,7 @@ async def _generate_fill_reports(self, command):
status = row["status"] if cum == filled else "partially_filled"
result = sink_observation(dict(row, status=status, filled_qty=str(cum),
filled_avg_price=str(average), updated_at_ns=execution["ts"],
execution_time_ns=execution["ts"], # FILL transaction_time, not startup order.updated_at
event="fill" if status == "filled" else "partial_fill",
execution_id=execution["trade_id"], event_qty=str(execution["qty"]),
event_price=str(execution["price"])))
Expand Down
262 changes: 188 additions & 74 deletions blueprints/us-equities/adaptive-paper/safety.py

Large diffs are not rendered by default.

12 changes: 6 additions & 6 deletions blueprints/us-equities/adaptive-paper/source-hashes.json
Original file line number Diff line number Diff line change
Expand Up @@ -21,21 +21,21 @@
"blueprints/us-equities/adaptive-paper/mover_runner.py": "ff1baf36baf4dd319cd0ed549cc5258194536548a098892bd203f8ab61c13bcd",
"blueprints/us-equities/adaptive-paper/mover_simulation.py": "8ac5955de1e814bb26fbfbca068a5719b9033b2bae0d8e50e78d7a78a546761e",
"blueprints/us-equities/adaptive-paper/mover_strategy.py": "4730c5e535aa480030d6484669a9d8565ca25e4e1ea50ea12a45bd1892536cba",
"blueprints/us-equities/adaptive-paper/native_adapter.py": "fe544c95ea2ee066052e1009e2f75a90e00ac5aaa34304f0264a00e5bc313b75",
"blueprints/us-equities/adaptive-paper/native_adapter.py": "fbf530884c2ad3b9e0c2b1f989eda399531647cc1db3c56dc9bcac10174fea47",
"blueprints/us-equities/adaptive-paper/native_strategy.py": "bc102eb6c4456cb651510fe59cf5f92decf3b70b31e0da1c833e7a27d6f1a226",
"blueprints/us-equities/adaptive-paper/receipts/adaptive_policy_v1.golden.json": "5cb4b21c7d2159e496093c67b3b358810db4e225d9c5ee1c633258050973556e",
"blueprints/us-equities/adaptive-paper/recovery.py": "5cf7dcea572efbda6979a2726eebf6d676f6bd62263900c7b6626da8b45c1215",
"blueprints/us-equities/adaptive-paper/registry.json": "60da73b985e718bd1dece7a3350f34636bda0fb38e766a44ae71c5a8c532282a",
"blueprints/us-equities/adaptive-paper/requirements.txt": "518afe4b77f901e87abd3a5d6faa126aa7e211db77eecec80cfe7ef42d82e07a",
"blueprints/us-equities/adaptive-paper/runner.py": "27bc3fdc80472c3ee218d3723e506288ebc6eef536dafc8365b30474213a6ae4",
"blueprints/us-equities/adaptive-paper/safety.py": "dd4ef862d9f8658a786db7f5df0fb0918ed177f01bad88db0e5029d535127626",
"blueprints/us-equities/adaptive-paper/safety.py": "c2d44faa1e4d369b8882dec9782ef32174c404b723edd2419833146b24eb2357",
"blueprints/us-equities/adaptive-paper/scheduled_trial.sh": "58a5b656875bda629443768cd0c69a4be28cbf90dfc9f5ae8782d7dcf3ac7d76",
"blueprints/us-equities/adaptive-paper/selector.py": "28de8ebf66b6e7135979554f73692ede772167682007d8c6ba41da6fb6fb39b3",
"blueprints/us-equities/adaptive-paper/sessions.py": "95f0affee512e88dd44f196fa72bb7471e00899356793bdbb12f15c4ed84e64d",
"blueprints/us-equities/adaptive-paper/simulation.py": "0119e0efe3b498be3c415eee9c1a0be922c9b5b17e3b774aa7528668c4eb0b30",
"blueprints/us-equities/adaptive-paper/strategies.py": "2d488a4fdd21ad323ab07cf7fb147b3623fd6e38951abcf2fc59cf8094a1383e",
"blueprints/us-equities/adaptive-paper/strategies_v1.py": "1952be269694485e12e4d927f124adca0ee29667588d6ac6651caede454c92f7",
"blueprints/us-equities/adaptive-paper/transport.py": "76c8a83d327545b2dd88a0bed416d857066d5389debaeddfce4137cc8ed5e484",
"blueprints/us-equities/adaptive-paper/transport.py": "e3038bd5562fecc7cea52df47c4ff6b94ddae7e85ce545d6684e651b856b89d9",
"tests/_credential_mutation_driver.py": "bfe35e69702016718452ed702aab20a0a33da5723e28a61de513b3c268849301",
"tests/adaptive_paper_hermetic.py": "0f41601cd10abf10ccba4f17d9c35716bb766a5e9951a4390649d6c9387fc1da",
"tests/test_adaptive_market_research.py": "772b4fd467d4405bd098a440c54de4bfd61f91bbb80693c052077b867f9e2fd6",
Expand All @@ -48,12 +48,12 @@
"tests/test_adaptive_paper_metrics.py": "1d6eaf940021fbaba3ac687a3d3a2d9f6e5492e85a5a372621299d4363ec3ee5",
"tests/test_adaptive_paper_mover.py": "0bb27aa6d5fb1883fb5e9b26179cf615c6ba589d2c46bb4898a1d7db50c1c0d3",
"tests/test_adaptive_paper_mover_native.py": "15ad37b5901ffac27bffcc1bc3b91f3b04307781de0103f45634b2046c5a9542",
"tests/test_adaptive_paper_native.py": "0713d85e034cc337b07d8bb889f18b32179420d843c264158901e260045191d6",
"tests/test_adaptive_paper_native.py": "728458624788243abcaf653f6a20a5a6201cfa9ec3328e8d7eb3fd58fdfa94b0",
"tests/test_adaptive_paper_recovery.py": "08e351ff903d6d008b802dc93a8dde07da8900df5cc0540f551ffa909cf4ce95",
"tests/test_adaptive_paper_runner.py": "29c4c48a8103cadb7906719d7145a9b14192b7d35486bb9eaa934faad5eac0b7",
"tests/test_adaptive_paper_safety.py": "a9287c503faf44cc9d6a5d824fb8cf39c545ff499221bde25d276a4af1aceae0",
"tests/test_adaptive_paper_safety.py": "68449ba3b2d4164c3c2c45c4c08cf7c51b171fce43e1ce0a0cedf08f32149235",
"tests/test_adaptive_paper_selector.py": "27b431ac7911372d050f25d32ac24a3df9d2cf3c5e7dafdebfe51db6cab55936",
"tests/test_adaptive_paper_sessions.py": "1e8b33ec8b496807468067112f5b7f0a2a6d6ce897ac7412605678b94984fc6a",
"tests/test_adaptive_paper_strategies.py": "f0814b36909f2fc6b4c4213b4142ffc13339740f0397c99ad9c7c81982e2033a",
"tests/test_adaptive_paper_transport.py": "6a9a17afa9aaee7ffab1625133655c9916e341a45f21e5a020ef3a027e9860ae"
"tests/test_adaptive_paper_transport.py": "5899a2498819d66db937e90e4a7fcec862f57fe8c551d8a5ca1f3dbc9e6e3a64"
}
15 changes: 11 additions & 4 deletions blueprints/us-equities/adaptive-paper/transport.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@
ACTIVITY_PAGE_SIZE = 100
# trade_updates events that carry one execution's own qty, price and execution_id.
EXECUTION_EVENTS = frozenset({"fill", "partial_fill"})
EXECUTION_FIELDS = ("event", "execution_id", "event_qty", "event_price")
EXECUTION_FIELDS = ("event", "execution_id", "event_qty", "event_price", "execution_time_ns")
# The pre-submission boundary (blueprints/us-equities/order-contract). Loaded by
# path at import, so a missing or broken contract fails transport import closed.
ORDER_CONTRACT_PATH = Path(__file__).resolve().parent.parent / "order-contract" / "order_contract.py"
Expand Down Expand Up @@ -328,11 +328,13 @@ def timestamp_ns(value):
raise TransportError("timestamp lacks timezone")
return int(value.timestamp()) * 1_000_000_000 + value.microsecond * 1000
if isinstance(value, str):
match = re.fullmatch(r"(.+?)(?:\.(\d{1,9}))?(Z|[+-]\d\d:\d\d)", value)
# RFC 3339 section 5.6 permits lowercase z and arbitrary fractional
# precision. Keep the first nine digits for integer nanoseconds.
match = re.fullmatch(r"(.+?)(?:\.(\d+))?([Zz]|[+-]\d\d:\d\d)", value)
if not match:
raise TransportError("invalid timestamp")
base = datetime.fromisoformat(match[1] + match[3].replace("Z", "+00:00"))
return int(base.timestamp()) * 1_000_000_000 + int((match[2] or "").ljust(9, "0"))
base = datetime.fromisoformat(match[1] + match[3].upper().replace("Z", "+00:00"))
return int(base.timestamp()) * 1_000_000_000 + int((match[2] or "")[:9].ljust(9, "0"))
raise TransportError("unsupported timestamp")


Expand Down Expand Up @@ -1290,6 +1292,11 @@ async def _consume_events(self):
("event_qty", "qty"), ("event_price", "price")):
if payload.get(source) is not None:
order[key] = str(payload[source])
# https://docs.alpaca.markets/docs/websocket-streaming:
# fill/partial_fill timestamp is when the execution occurred;
# nested order.updated_at is a separate order-state timestamp.
if order.get("event") in EXECUTION_EVENTS and payload.get("timestamp") is not None:
order["execution_time_ns"] = timestamp_ns(payload["timestamp"])
self._stream_seen.add(order["client_order_id"])
self._pending_stream.pop(order["client_order_id"], None)
order = await self._observe(order)
Expand Down
26 changes: 13 additions & 13 deletions manifests/evidence.json
Original file line number Diff line number Diff line change
Expand Up @@ -9773,8 +9773,8 @@
},
{
"path": "blueprints/us-equities/adaptive-paper/native_adapter.py",
"sha256": "fe544c95ea2ee066052e1009e2f75a90e00ac5aaa34304f0264a00e5bc313b75",
"bytes": 53661
"sha256": "fbf530884c2ad3b9e0c2b1f989eda399531647cc1db3c56dc9bcac10174fea47",
"bytes": 54033
},
{
"path": "blueprints/us-equities/adaptive-paper/native_strategy.py",
Expand Down Expand Up @@ -9813,8 +9813,8 @@
},
{
"path": "blueprints/us-equities/adaptive-paper/safety.py",
"sha256": "dd4ef862d9f8658a786db7f5df0fb0918ed177f01bad88db0e5029d535127626",
"bytes": 75461
"sha256": "c2d44faa1e4d369b8882dec9782ef32174c404b723edd2419833146b24eb2357",
"bytes": 83006
},
{
"path": "blueprints/us-equities/adaptive-paper/selector.py",
Expand All @@ -9833,7 +9833,7 @@
},
{
"path": "blueprints/us-equities/adaptive-paper/source-hashes.json",
"sha256": "9f5e65c472a214a85dd5657260e9a7aaca3075bd5c0161a95a7ded2658759f2d",
"sha256": "f2832e37855426ff7e296cb085d60b640fb20375e2e1527a715ac734c5fdbf98",
"bytes": 6969
},
{
Expand All @@ -9848,8 +9848,8 @@
},
{
"path": "blueprints/us-equities/adaptive-paper/transport.py",
"sha256": "76c8a83d327545b2dd88a0bed416d857066d5389debaeddfce4137cc8ed5e484",
"bytes": 82686
"sha256": "e3038bd5562fecc7cea52df47c4ff6b94ddae7e85ce545d6684e651b856b89d9",
"bytes": 83300
},
{
"path": "blueprints/us-equities/adaptive-paper/trials/20260923-interrupted/README.md",
Expand Down Expand Up @@ -37058,8 +37058,8 @@
},
{
"path": "tests/test_adaptive_paper_native.py",
"sha256": "0713d85e034cc337b07d8bb889f18b32179420d843c264158901e260045191d6",
"bytes": 83385
"sha256": "728458624788243abcaf653f6a20a5a6201cfa9ec3328e8d7eb3fd58fdfa94b0",
"bytes": 85779
},
{
"path": "tests/test_adaptive_paper_recovery.py",
Expand All @@ -37073,8 +37073,8 @@
},
{
"path": "tests/test_adaptive_paper_safety.py",
"sha256": "a9287c503faf44cc9d6a5d824fb8cf39c545ff499221bde25d276a4af1aceae0",
"bytes": 75302
"sha256": "68449ba3b2d4164c3c2c45c4c08cf7c51b171fce43e1ce0a0cedf08f32149235",
"bytes": 94507
},
{
"path": "tests/test_adaptive_paper_sessions.py",
Expand All @@ -37088,8 +37088,8 @@
},
{
"path": "tests/test_adaptive_paper_transport.py",
"sha256": "6a9a17afa9aaee7ffab1625133655c9916e341a45f21e5a020ef3a027e9860ae",
"bytes": 94796
"sha256": "5899a2498819d66db937e90e4a7fcec862f57fe8c551d8a5ca1f3dbc9e6e3a64",
"bytes": 103152
},
{
"path": "tests/test_adoption_bootstrap.py",
Expand Down
39 changes: 39 additions & 0 deletions tests/test_adaptive_paper_native.py
Original file line number Diff line number Diff line change
Expand Up @@ -300,6 +300,13 @@ async def exercise():
self.assertEqual(reopened.positions()["SPY"].cost_basis_usd, Decimal("300.02"))
self.assertEqual(reopened.accounting().cash_delta_usd, Decimal("-300.02"))
self.assertEqual(reopened.unresolved(), [])
# https://docs.alpaca.markets/docs/account-activities: FILL's
# transaction_time is the execution time. Adjacent nanoseconds
# must survive the recovery hook, Controller and SQLite reopen.
executions = reopened.db.execute("SELECT * FROM executions ORDER BY cum_qty").fetchall()
self.assertIn("execution_time_ns", executions[0].keys())
self.assertEqual([r["execution_time_ns"] for r in executions],
[r["transaction_time_ns"] for r in port.activities])

def test_real_native_roundtrip_partial_and_duplicate_events(self):
port, strategy, session = self.run_node("fills")
Expand Down Expand Up @@ -1190,6 +1197,24 @@ def test_startup_reports_survive_reopen_without_later_observation(self):
self.assertEqual(self.executions(), self.expected_executions())
self.assert_exact_accounting()

def test_startup_execution_times_are_durable_before_any_later_observation(self):
# Startup has a separate _adopted_executions -> sink_observation loop.
# Alpaca FILL transaction_time (account-activities) and rc5 FillReport
# must describe the same execution, including the last nanosecond.
expected = {}
for activities in self.activities.values():
for index, activity in enumerate(activities):
stamp = TRANSPORT.timestamp_ns("2026-09-24T15:00:21.12345678%dZ" % index)
activity["transaction_time_ns"] = stamp
expected[activity["trade_id"]] = stamp
reports = self.reports()
self.reopen()
rows = self.ledger.db.execute("SELECT * FROM executions WHERE client_id != 'held-buy'").fetchall()
self.assertEqual(len(rows), 4)
self.assertIn("execution_time_ns", rows[0].keys())
self.assertEqual({r["execution_id"]: r["execution_time_ns"] for r in rows}, expected)
self.assertEqual({str(r.trade_id): r.ts_event for r in reports}, expected)

def test_restart_does_not_double_book_previously_recorded_startup_fills(self):
self.deliver(self.deliveries)
self.assert_exact_accounting()
Expand All @@ -1203,6 +1228,20 @@ def test_restart_does_not_double_book_previously_recorded_startup_fills(self):
self.assertEqual(self.durable_state(), before)
self.assertEqual(self.executions(), self.expected_executions())

def test_startup_redelivery_keeps_untimed_executions_in_journal_positions(self):
# probe_startup_native.py: an activity can have a transaction time
# even though its already-booked stream/legacy execution was untimed.
self.deliver(self.deliveries)
before = tuple(self.ledger.db.iterdump())
reports = self.reports()
self.assertEqual(len(reports), 4)
self.assertTrue(all(report.ts_event > 0 for report in reports))
self.reopen()
times = [row[0] for row in self.ledger.db.execute(
"SELECT execution_time_ns FROM executions WHERE client_id != 'held-buy'")]
self.assertEqual(times, [None] * 4)
self.assertEqual(tuple(self.ledger.db.iterdump()), before)

def test_startup_report_quantities_and_prices_match_durable_executions(self):
reports = self.reports()
reported = {(str(r.client_order_id), str(r.trade_id)):
Expand Down
Loading
Loading