Spaces:
Sleeping
Sleeping
File size: 2,488 Bytes
c7ee404 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 | from __future__ import annotations
from datetime import datetime, timezone
from typing import Any
from sqlalchemy import Boolean, DateTime, Float, Integer, String
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.types import JSON
JsonColumn = JSON().with_variant(JSONB, "postgresql")
def utcnow() -> datetime:
return datetime.now(timezone.utc)
class Base(DeclarativeBase):
pass
class PredictionLog(Base):
__tablename__ = "prediction_logs"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
timestamp: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
event: Mapped[str] = mapped_column(String(40), nullable=False, default="prediction")
input_data: Mapped[dict[str, Any]] = mapped_column(JsonColumn, nullable=False)
prediction: Mapped[int | None] = mapped_column(Integer, nullable=True)
probability_of_default: Mapped[float | None] = mapped_column(Float, nullable=True)
risk_category: Mapped[str | None] = mapped_column(String(20), nullable=True)
inference_time_ms: Mapped[float | None] = mapped_column(Float, nullable=True)
error: Mapped[str | None] = mapped_column(String, nullable=True)
model_version: Mapped[str | None] = mapped_column(String(120), nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
nullable=False,
default=utcnow,
)
class DriftEvent(Base):
__tablename__ = "drift_events"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
timestamp: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
drift_detected: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
feature: Mapped[str | None] = mapped_column(String(100), nullable=True)
drift_score: Mapped[float | None] = mapped_column(Float, nullable=True)
p_value: Mapped[float | None] = mapped_column(Float, nullable=True)
ks_statistic: Mapped[float | None] = mapped_column(Float, nullable=True)
method: Mapped[str | None] = mapped_column(String(50), nullable=True)
batch_size: Mapped[int | None] = mapped_column(Integer, nullable=True)
details: Mapped[dict[str, Any] | None] = mapped_column(JsonColumn, nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
nullable=False,
default=utcnow,
)
|