Skip to content

Carry snapshot diagnostics into execution #116

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Pigbibi merged 1 commit into from
Jun 1, 2026
+67 −0
Merged

Carry snapshot diagnostics into execution #116

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
32 changes: 32 additions & 0 deletions decision_mapper.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,17 @@
_INCOME_SYMBOLS = frozenset({"QQQI", "SPYI"})
_DEFAULT_MIN_TRADE_FLOOR = 100.0
_DEFAULT_REBALANCE_THRESHOLD_RATIO = 0.01
_SNAPSHOT_DIAGNOSTIC_FIELDS = (
"snapshot_manifest_price_as_of",
"snapshot_manifest_universe_as_of",
"snapshot_manifest_source_input_status",
"snapshot_manifest_source_input_fallback_used",
"snapshot_manifest_source_input_fallback_reason",
"snapshot_manifest_source_input_fallback_streak",
"snapshot_manifest_source_input_manifest_path",
"snapshot_manifest_source_refresh_run_id",
"snapshot_manifest_source_refresh_generated_at",
)


def _build_portfolio_inputs(
Expand Down Expand Up @@ -106,6 +117,22 @@ def _resolve_platform_reserved_cash(
return max(reserved_cash_floor_usd, max(0.0, float(total_equity)) * reserved_cash_ratio)


def _attach_snapshot_diagnostics(
plan: dict[str, Any],
*,
decision: StrategyDecision,
runtime_metadata: Mapping[str, Any] | None,
) -> None:
execution = plan.get("execution")
if not isinstance(execution, dict):
return
diagnostics = {**dict(runtime_metadata or {}), **dict(decision.diagnostics)}
for field in _SNAPSHOT_DIAGNOSTIC_FIELDS:
value = diagnostics.get(field)
if value is not None and value != "":
execution[field] = value


def _apply_reserved_cash_policy(
annotations: ValueTargetExecutionAnnotations,
*,
Expand Down Expand Up @@ -498,4 +525,9 @@ def map_strategy_decision_to_plan(
cash_by_currency = _cash_by_currency_from_snapshot(snapshot)
if cash_by_currency:
plan["portfolio"]["cash_by_currency"] = cash_by_currency
_attach_snapshot_diagnostics(
plan,
decision=normalized_decision,

@chatgpt-codex-connector chatgpt-codex-connector Bot Jun 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Preserve original decision diagnostics when attaching snapshot data

When a no_execute decision carries snapshot manifest fields in decision.diagnostics, _normalize_to_value_target_decision replaces it with a synthetic hold-current decision whose diagnostics are empty, so passing normalized_decision here drops the very fields this helper tries to copy into execution. In that scenario the signal snapshot/notification still won't show the manifest dates or fallback status unless the same values also happen to be duplicated in runtime_metadata; use the original decision (or merge its diagnostics before normalization) for this attachment step.

Useful? React with 👍 / 👎.

runtime_metadata=runtime_metadata,
)
return plan
35 changes: 35 additions & 0 deletions tests/test_decision_mapper.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -212,6 +212,41 @@ def test_applies_platform_reserved_cash_policy_to_weight_decision(self):
self.assertEqual(plan["execution"]["reserved_cash"], 1500.0)
self.assertEqual(plan["execution"]["investable_cash"], 2500.0)

def test_carries_snapshot_manifest_diagnostics_to_execution(self):
decision = StrategyDecision(
positions=(),
risk_flags=("no_execute",),
diagnostics={"signal_description": "monthly cadence"},
)
snapshot = PortfolioSnapshot(
as_of=datetime.now(timezone.utc),
total_equity=10000.0,
buying_power=10000.0,
positions=(),
metadata={"account_hash": "longbridge-snapshot"},
)

plan = map_strategy_decision_to_plan(
decision,
snapshot=snapshot,
strategy_profile="mega_cap_leader_rotation_top50_balanced",
runtime_metadata={
"snapshot_manifest_price_as_of": "2026-06-01",
"snapshot_manifest_universe_as_of": "2026-04-29",
"snapshot_manifest_source_input_status": "universe_fallback",
"snapshot_manifest_source_input_fallback_used": True,
"snapshot_manifest_source_input_fallback_streak": 1,
"snapshot_manifest_source_refresh_run_id": "26785047433",
},
)

self.assertEqual(plan["execution"]["snapshot_manifest_price_as_of"], "2026-06-01")
self.assertEqual(plan["execution"]["snapshot_manifest_universe_as_of"], "2026-04-29")
self.assertEqual(plan["execution"]["snapshot_manifest_source_input_status"], "universe_fallback")
self.assertIs(plan["execution"]["snapshot_manifest_source_input_fallback_used"], True)
self.assertEqual(plan["execution"]["snapshot_manifest_source_input_fallback_streak"], 1)
self.assertEqual(plan["execution"]["snapshot_manifest_source_refresh_run_id"], "26785047433")

def test_platform_reserved_cash_policy_does_not_lower_strategy_reserve(self):
decision = StrategyDecision(
positions=(PositionTarget(symbol="TQQQ", target_value=5000.0),),
Expand Down