gavindiaz/polymarket-insider-tracker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
jp-vps-deploy
2026-05-23 02:13:33 +00:00
parent ff678468e3
commit 8a0e8c9915
6 changed files with 331 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
alembic/versions/20260522_1130_risk_assessments.py
+64
View File
@@ -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
+26
View File
@@ -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
+6 -2
View File
@@ -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:"
src/polymarket_insider_tracker/pipeline.py
+74 -2
View File
@@ -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
+54
View File
@@ -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
+107
View File
@@ -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,
)