diff --git a/services/kevin_signal_bridge/audit.py b/services/kevin_signal_bridge/audit.py index 5c7b1c0..ae53320 100644 --- a/services/kevin_signal_bridge/audit.py +++ b/services/kevin_signal_bridge/audit.py @@ -9,12 +9,36 @@ from typing import Any, Callable from sqlalchemy.dialects.postgresql import insert as pg_insert from shared.models.meet_kevin_trading import KevinSignalBridgeState +from shared.models.trading import Signal as SignalRow +from shared.schemas.trading import TradeSignal class AuditWriter: def __init__(self, session_factory: Callable[..., Any]) -> None: self.session_factory = session_factory + async def persist_signal(self, signal: TradeSignal) -> None: + """Persist a TradeSignal to the `signals` table so downstream + FKs (kevin_signal_bridge_state.signal_id, trades.signal_id) + resolve. Idempotent on the signal UUID — caller should call + once per published signal before writing the audit row. + """ + async with self.session_factory() as session: + stmt = pg_insert(SignalRow).values( + id=signal.signal_id, + ticker=signal.ticker, + direction=signal.direction, + strength=signal.strength, + strategy_sources={"sources": signal.strategy_sources}, + strategy_id=signal.strategy_id, + acted_on=False, + ) + # Idempotent — if the row already exists (e.g. retry path) + # leave it alone. + stmt = stmt.on_conflict_do_nothing(index_elements=["id"]) + await session.execute(stmt) + await session.commit() + async def write( self, *, diff --git a/services/kevin_signal_bridge/main.py b/services/kevin_signal_bridge/main.py index c011f42..2216d0f 100644 --- a/services/kevin_signal_bridge/main.py +++ b/services/kevin_signal_bridge/main.py @@ -131,6 +131,11 @@ class KevinBridge: take_profit_pct=Decimal(str(self.config.kevin_take_profit_pct)), ) + # Persist to signals table FIRST so downstream FK + # (kevin_signal_bridge_state.signal_id → signals.id) resolves. + # On retry the on_conflict_do_nothing makes this idempotent. + await self.audit_writer.persist_signal(signal) + try: stream_id = await self.publisher.publish(signal) except Exception: