diff --git a/alembic/versions/20260522_1130_risk_assessments.py b/alembic/versions/20260522_1130_risk_assessments.py new file mode 100644 index 0000000..f95aeb3 --- /dev/null +++ b/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") diff --git a/src/polymarket_insider_tracker/config.py b/src/polymarket_insider_tracker/config.py index 83a8d40..da862e4 100644 --- a/src/polymarket_insider_tracker/config.py +++ b/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( diff --git a/src/polymarket_insider_tracker/detector/scorer.py b/src/polymarket_insider_tracker/detector/scorer.py index 07691e5..982d48b 100644 --- a/src/polymarket_insider_tracker/detector/scorer.py +++ b/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_ALERT_THRESHOLD = 0.6 +# Default configuration. The threshold lifted from 0.6 to 0.80 after the +# 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:" diff --git a/src/polymarket_insider_tracker/pipeline.py b/src/polymarket_insider_tracker/pipeline.py index fb27bcb..6e48802 100644 --- a/src/polymarket_insider_tracker/pipeline.py +++ b/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. diff --git a/src/polymarket_insider_tracker/storage/models.py b/src/polymarket_insider_tracker/storage/models.py index eb8bace..7a0322a 100644 --- a/src/polymarket_insider_tracker/storage/models.py +++ b/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"), + ) diff --git a/src/polymarket_insider_tracker/storage/repos.py b/src/polymarket_insider_tracker/storage/repos.py index a9b7bdf..65ad6d6 100644 --- a/src/polymarket_insider_tracker/storage/repos.py +++ b/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, + )