feat(detector): persist all risk assessments to risk_assessments table

alembic/versions/20260522_1130_risk_assessments.py

@@ -0,0 +1,64 @@
+"""Risk assessment persistence layer.
+
+Adds the `risk_assessments` table — one row per signal-bearing trade —
+so future backtests can rebuild ground truth without grepping the
+systemd log or hammering the public data-api.
+
+Revision ID: 002_risk_assessments
+Revises: 001_initial
+Create Date: 2026-05-22 11:30:00.000000+00:00
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "002_risk_assessments"
+down_revision: str | None = "001_initial"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "risk_assessments",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("assessment_id", sa.String(36), nullable=False),
+        sa.Column("trade_id", sa.String(80), nullable=False),
+        sa.Column("wallet_address", sa.String(42), nullable=False),
+        sa.Column("market_id", sa.String(80), nullable=False),
+        sa.Column("asset_id", sa.String(80), nullable=True),
+        sa.Column("side", sa.String(8), nullable=False),
+        sa.Column("outcome", sa.String(120), nullable=True),
+        sa.Column("outcome_index", sa.Integer(), nullable=True),
+        sa.Column("price", sa.Numeric(10, 6), nullable=False),
+        sa.Column("size", sa.Numeric(20, 6), nullable=False),
+        sa.Column("notional_usdc", sa.Numeric(20, 6), nullable=False),
+        sa.Column("trade_timestamp", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("weighted_score", sa.Numeric(4, 3), nullable=False),
+        sa.Column("signals_triggered", sa.Integer(), nullable=False),
+        sa.Column("fresh_wallet_confidence", sa.Numeric(4, 3), nullable=True),
+        sa.Column("size_anomaly_confidence", sa.Numeric(4, 3), nullable=True),
+        sa.Column("is_niche_market", sa.Boolean(), nullable=True),
+        sa.Column("volume_impact", sa.Numeric(8, 4), nullable=True),
+        sa.Column("book_impact", sa.Numeric(8, 4), nullable=True),
+        sa.Column("wallet_age_hours", sa.Numeric(10, 2), nullable=True),
+        sa.Column("should_alert", sa.Boolean(), nullable=False),
+        sa.Column("threshold_at_eval", sa.Numeric(4, 3), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("assessment_id"),
+    )
+    op.create_index("idx_risk_assessments_wallet", "risk_assessments", ["wallet_address"])
+    op.create_index("idx_risk_assessments_market", "risk_assessments", ["market_id"])
+    op.create_index("idx_risk_assessments_trade_ts", "risk_assessments", ["trade_timestamp"])
+    op.create_index("idx_risk_assessments_score", "risk_assessments", ["weighted_score"])
+
+
+def downgrade() -> None:
+    op.drop_index("idx_risk_assessments_score", table_name="risk_assessments")
+    op.drop_index("idx_risk_assessments_trade_ts", table_name="risk_assessments")
+    op.drop_index("idx_risk_assessments_market", table_name="risk_assessments")
+    op.drop_index("idx_risk_assessments_wallet", table_name="risk_assessments")
+    op.drop_table("risk_assessments")

src/polymarket_insider_tracker/config.py

@@ -145,6 +145,31 @@ class TelegramSettings(BaseSettings):
         return self.bot_token is not None and bool(self.bot_token.get_secret_value().strip()) and self.chat_id is not None and bool(self.chat_id.strip())
 
 
+class DetectorSettings(BaseSettings):
+    """Risk-scorer / detector tuning."""
+
+    model_config = SettingsConfigDict(env_prefix="DETECTOR_", env_file=".env", env_file_encoding="utf-8", extra="ignore")
+
+    alert_threshold: float = Field(
+        default=0.80,
+        alias="DETECTOR_ALERT_THRESHOLD",
+        description="Minimum weighted score required to trigger an alert",
+        ge=0.0,
+        le=1.0,
+    )
+    dedup_window_seconds: int = Field(
+        default=3600,
+        alias="DETECTOR_DEDUP_WINDOW_SECONDS",
+        description="Per-(wallet, market) dedup window in seconds",
+        ge=0,
+    )
+    persist_assessments: bool = Field(
+        default=True,
+        alias="DETECTOR_PERSIST_ASSESSMENTS",
+        description="Write every signal-bearing risk assessment to the database",
+    )
+
+
 class Settings(BaseSettings):
     """Main application settings.
 
@@ -174,6 +199,7 @@ class Settings(BaseSettings):
     polymarket: PolymarketSettings = Field(default_factory=PolymarketSettings)
     discord: DiscordSettings = Field(default_factory=DiscordSettings)
     telegram: TelegramSettings = Field(default_factory=TelegramSettings)
+    detector: DetectorSettings = Field(default_factory=DetectorSettings)
 
     # Application settings
     log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = Field(

src/polymarket_insider_tracker/detector/scorer.py

@@ -19,8 +19,12 @@ from polymarket_insider_tracker.ingestor.models import TradeEvent
 
 logger = logging.getLogger(__name__)
 
-# Default configuration
+# Default configuration. The threshold lifted from 0.6 to 0.80 after the
-DEFAULT_ALERT_THRESHOLD = 0.6
+# first cost-adjusted backtest showed everything below 0.85 was follower-PnL
+# negative under realistic taker fees + half-cent slippage. 0.80 keeps a small
+# margin below 0.85+ so we don't drop borderline-high signals on a hard cliff.
+# Override at runtime via DETECTOR_ALERT_THRESHOLD env var.
+DEFAULT_ALERT_THRESHOLD = 0.80
 DEFAULT_DEDUP_WINDOW_SECONDS = 3600  # 1 hour
 DEFAULT_REDIS_KEY_PREFIX = "polymarket:dedup:"
 

src/polymarket_insider_tracker/pipeline.py

@@ -34,6 +34,8 @@ from polymarket_insider_tracker.storage.database import DatabaseManager
 from polymarket_insider_tracker.storage.repos import (
     FundingRepository,
     FundingTransferDTO,
+    RiskAssessmentDTO,
+    RiskAssessmentRepository,
     WalletProfileDTO,
     WalletRepository,
 )
@@ -43,6 +45,7 @@ if TYPE_CHECKING:
 
     from polymarket_insider_tracker.detector.models import (
         FreshWalletSignal,
+        RiskAssessment,
         SizeAnomalySignal,
     )
     from polymarket_insider_tracker.ingestor.models import TradeEvent
@@ -252,7 +255,17 @@ class Pipeline:
 
         # Initialize Risk Scorer
         logger.debug("Initializing risk scorer...")
-        self._risk_scorer = RiskScorer(self._redis)
+        self._risk_scorer = RiskScorer(
+            self._redis,
+            alert_threshold=settings.detector.alert_threshold,
+            dedup_window_seconds=settings.detector.dedup_window_seconds,
+        )
+        logger.info(
+            "RiskScorer threshold=%.2f dedup_window=%ds persist=%s",
+            settings.detector.alert_threshold,
+            settings.detector.dedup_window_seconds,
+            settings.detector.persist_assessments,
+        )
 
         # Initialize Alerting
         logger.debug("Initializing alerting components...")
@@ -476,13 +489,19 @@ class Pipeline:
             return None
 
     async def _score_and_alert(self, bundle: SignalBundle) -> None:
-        """Score signals and send alert if threshold exceeded."""
+        """Score signals, persist the assessment, and send alert if above threshold."""
         if not self._risk_scorer or not self._alert_formatter or not self._alert_dispatcher:
             return
 
         # Get risk assessment
         assessment = await self._risk_scorer.assess(bundle)
 
+        # Persist every signal-bearing assessment (not just delivered alerts).
+        # This is the ground-truth log future backtests will read instead of
+        # grepping systemd. Failure here must never block alerting.
+        if self._settings.detector.persist_assessments:
+            await self._persist_assessment(assessment)
+
         if not assessment.should_alert:
             logger.debug(
                 "Trade %s below alert threshold (score=%.2f)",
@@ -518,6 +537,59 @@ class Pipeline:
                 result.success_count + result.failure_count,
             )
 
+    async def _persist_assessment(self, assessment: "RiskAssessment") -> None:
+        """Write the assessment row. Best-effort; never raises."""
+        if not self._db_manager:
+            return
+        from decimal import Decimal as _D
+
+        trade = assessment.trade_event
+        fresh = assessment.fresh_wallet_signal
+        size_sig = assessment.size_anomaly_signal
+        wallet_age: _D | None = None
+        if fresh is not None and fresh.wallet_profile.age_hours is not None:
+            wallet_age = _D(str(round(float(fresh.wallet_profile.age_hours), 2)))
+        dto = RiskAssessmentDTO(
+            assessment_id=assessment.assessment_id,
+            trade_id=trade.trade_id,
+            wallet_address=assessment.wallet_address.lower(),
+            market_id=assessment.market_id,
+            asset_id=getattr(trade, "asset_id", None) or None,
+            side=trade.side,
+            outcome=getattr(trade, "outcome", None) or None,
+            outcome_index=getattr(trade, "outcome_index", None),
+            price=trade.price,
+            size=trade.size,
+            notional_usdc=trade.notional_value,
+            trade_timestamp=trade.timestamp,
+            weighted_score=_D(str(round(assessment.weighted_score, 3))),
+            signals_triggered=assessment.signals_triggered,
+            fresh_wallet_confidence=(
+                _D(str(round(fresh.confidence, 3))) if fresh is not None else None
+            ),
+            size_anomaly_confidence=(
+                _D(str(round(size_sig.confidence, 3))) if size_sig is not None else None
+            ),
+            is_niche_market=size_sig.is_niche_market if size_sig is not None else None,
+            volume_impact=(
+                _D(str(round(size_sig.volume_impact, 4))) if size_sig is not None else None
+            ),
+            book_impact=(
+                _D(str(round(size_sig.book_impact, 4))) if size_sig is not None else None
+            ),
+            wallet_age_hours=wallet_age,
+            should_alert=assessment.should_alert,
+            threshold_at_eval=_D(str(round(self._settings.detector.alert_threshold, 3))),
+        )
+        try:
+            async with self._db_manager.get_async_session() as session:
+                repo = RiskAssessmentRepository(session)
+                await repo.insert(dto)
+        except Exception as e:
+            logger.warning(
+                "Failed to persist risk assessment %s: %s", assessment.assessment_id, e
+            )
+
     async def run(self) -> None:
         """Start the pipeline and run until interrupted.
 

src/polymarket_insider_tracker/storage/models.py

@@ -115,3 +115,57 @@ class WalletRelationshipModel(Base):
         Index("idx_wallet_relationships_a", "wallet_a"),
         Index("idx_wallet_relationships_b", "wallet_b"),
     )
+
+
+class RiskAssessmentModel(Base):
+    """SQLAlchemy model for risk assessments.
+
+    One row per signal-bearing trade (i.e. trades that triggered at least one
+    detector). Captures everything a future backtest needs without going back
+    to the public API: trade identity, score, per-signal confidences, and
+    whether the alert was actually delivered (could be False due to dedup or
+    threshold).
+    """
+
+    __tablename__ = "risk_assessments"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    assessment_id: Mapped[str] = mapped_column(String(36), unique=True, nullable=False)
+
+    # Trade identity
+    trade_id: Mapped[str] = mapped_column(String(80), nullable=False)
+    wallet_address: Mapped[str] = mapped_column(String(42), nullable=False)
+    market_id: Mapped[str] = mapped_column(String(80), nullable=False)
+    asset_id: Mapped[str | None] = mapped_column(String(80), nullable=True)
+    side: Mapped[str] = mapped_column(String(8), nullable=False)
+    outcome: Mapped[str | None] = mapped_column(String(120), nullable=True)
+    outcome_index: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    price: Mapped[Decimal] = mapped_column(Numeric(10, 6), nullable=False)
+    size: Mapped[Decimal] = mapped_column(Numeric(20, 6), nullable=False)
+    notional_usdc: Mapped[Decimal] = mapped_column(Numeric(20, 6), nullable=False)
+    trade_timestamp: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+
+    # Scoring
+    weighted_score: Mapped[Decimal] = mapped_column(Numeric(4, 3), nullable=False)
+    signals_triggered: Mapped[int] = mapped_column(Integer, nullable=False)
+    fresh_wallet_confidence: Mapped[Decimal | None] = mapped_column(Numeric(4, 3), nullable=True)
+    size_anomaly_confidence: Mapped[Decimal | None] = mapped_column(Numeric(4, 3), nullable=True)
+    is_niche_market: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
+    volume_impact: Mapped[Decimal | None] = mapped_column(Numeric(8, 4), nullable=True)
+    book_impact: Mapped[Decimal | None] = mapped_column(Numeric(8, 4), nullable=True)
+    wallet_age_hours: Mapped[Decimal | None] = mapped_column(Numeric(10, 2), nullable=True)
+
+    # Decision
+    should_alert: Mapped[bool] = mapped_column(Boolean, nullable=False)
+    threshold_at_eval: Mapped[Decimal] = mapped_column(Numeric(4, 3), nullable=False)
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC)
+    )
+
+    __table_args__ = (
+        Index("idx_risk_assessments_wallet", "wallet_address"),
+        Index("idx_risk_assessments_market", "market_id"),
+        Index("idx_risk_assessments_trade_ts", "trade_timestamp"),
+        Index("idx_risk_assessments_score", "weighted_score"),
+    )

src/polymarket_insider_tracker/storage/repos.py

@@ -18,6 +18,7 @@ from sqlalchemy.dialects.sqlite import insert as sqlite_insert
 
 from polymarket_insider_tracker.storage.models import (
     FundingTransferModel,
+    RiskAssessmentModel,
     WalletProfileModel,
     WalletRelationshipModel,
 )
@@ -510,3 +511,109 @@ class RelationshipRepository:
         )
         # SQLAlchemy Result does have rowcount but typing doesn't reflect it
         return (result.rowcount or 0) > 0  # type: ignore[attr-defined]
+
+
+@dataclass
+class RiskAssessmentDTO:
+    """Data transfer object for a persisted risk assessment.
+
+    Captures everything a future backtest needs without going back to
+    public APIs: trade identity, score, per-signal confidences, and
+    whether the alert was actually delivered.
+    """
+
+    assessment_id: str
+    trade_id: str
+    wallet_address: str
+    market_id: str
+    asset_id: str | None
+    side: str
+    outcome: str | None
+    outcome_index: int | None
+    price: Decimal
+    size: Decimal
+    notional_usdc: Decimal
+    trade_timestamp: datetime
+    weighted_score: Decimal
+    signals_triggered: int
+    fresh_wallet_confidence: Decimal | None
+    size_anomaly_confidence: Decimal | None
+    is_niche_market: bool | None
+    volume_impact: Decimal | None
+    book_impact: Decimal | None
+    wallet_age_hours: Decimal | None
+    should_alert: bool
+    threshold_at_eval: Decimal
+    created_at: datetime | None = None
+
+
+class RiskAssessmentRepository:
+    """Repository for risk assessment data access."""
+
+    def __init__(self, session: AsyncSession) -> None:
+        self.session = session
+
+    async def insert(self, dto: RiskAssessmentDTO) -> RiskAssessmentDTO:
+        """Insert a single assessment. Idempotent on assessment_id collisions."""
+        model = RiskAssessmentModel(
+            assessment_id=dto.assessment_id,
+            trade_id=dto.trade_id,
+            wallet_address=dto.wallet_address.lower(),
+            market_id=dto.market_id,
+            asset_id=dto.asset_id,
+            side=dto.side,
+            outcome=dto.outcome,
+            outcome_index=dto.outcome_index,
+            price=dto.price,
+            size=dto.size,
+            notional_usdc=dto.notional_usdc,
+            trade_timestamp=dto.trade_timestamp,
+            weighted_score=dto.weighted_score,
+            signals_triggered=dto.signals_triggered,
+            fresh_wallet_confidence=dto.fresh_wallet_confidence,
+            size_anomaly_confidence=dto.size_anomaly_confidence,
+            is_niche_market=dto.is_niche_market,
+            volume_impact=dto.volume_impact,
+            book_impact=dto.book_impact,
+            wallet_age_hours=dto.wallet_age_hours,
+            should_alert=dto.should_alert,
+            threshold_at_eval=dto.threshold_at_eval,
+        )
+        self.session.add(model)
+        await self.session.flush()
+        return dto
+
+    async def get_by_assessment_id(self, assessment_id: str) -> RiskAssessmentDTO | None:
+        result = await self.session.execute(
+            select(RiskAssessmentModel).where(
+                RiskAssessmentModel.assessment_id == assessment_id
+            )
+        )
+        model = result.scalar_one_or_none()
+        if model is None:
+            return None
+        return RiskAssessmentDTO(
+            assessment_id=model.assessment_id,
+            trade_id=model.trade_id,
+            wallet_address=model.wallet_address,
+            market_id=model.market_id,
+            asset_id=model.asset_id,
+            side=model.side,
+            outcome=model.outcome,
+            outcome_index=model.outcome_index,
+            price=model.price,
+            size=model.size,
+            notional_usdc=model.notional_usdc,
+            trade_timestamp=model.trade_timestamp,
+            weighted_score=model.weighted_score,
+            signals_triggered=model.signals_triggered,
+            fresh_wallet_confidence=model.fresh_wallet_confidence,
+            size_anomaly_confidence=model.size_anomaly_confidence,
+            is_niche_market=model.is_niche_market,
+            volume_impact=model.volume_impact,
+            book_impact=model.book_impact,
+            wallet_age_hours=model.wallet_age_hours,
+            should_alert=model.should_alert,
+            threshold_at_eval=model.threshold_at_eval,
+            created_at=model.created_at,
+        )