feat: add training case engine M1 persistence layer

This commit is contained in:
2026-04-30 19:53:31 -04:00
parent 87f6f35c7c
commit 5e7e3681cf
7 changed files with 841 additions and 1 deletions
@@ -0,0 +1,119 @@
create table if not exists training_cases (
id bigserial primary key,
instrument_id bigint not null references instruments(id),
timeframe text not null,
case_type text not null,
skill_tag text not null,
chapter_ref text,
source_event_type text,
source_event_id bigint,
source_event_refs jsonb not null default '[]'::jsonb,
start_ts timestamptz not null,
decision_ts timestamptz not null,
end_ts timestamptz not null,
difficulty text not null default 'medium',
answer_key jsonb not null default '{}'::jsonb,
status text not null default 'draft',
meta jsonb not null default '{}'::jsonb,
created_at timestamptz not null default now(),
updated_at timestamptz not null default now(),
check (start_ts <= decision_ts),
check (decision_ts <= end_ts)
);
create index if not exists idx_training_cases_instr_tf_decision_desc
on training_cases (instrument_id, timeframe, decision_ts desc);
create index if not exists idx_training_cases_skill_status_created_desc
on training_cases (skill_tag, status, created_at desc);
create index if not exists idx_training_cases_type_status_created_desc
on training_cases (case_type, status, created_at desc);
create table if not exists training_case_tasks (
id bigserial primary key,
case_id bigint not null references training_cases(id) on delete cascade,
task_type text not null,
skill_tag text not null,
prompt text not null,
required_annotation_types jsonb not null default '[]'::jsonb,
max_score numeric(8,2) not null default 0,
sort_order integer not null default 0,
status text not null default 'active',
meta jsonb not null default '{}'::jsonb,
created_at timestamptz not null default now(),
updated_at timestamptz not null default now()
);
create index if not exists idx_training_case_tasks_case_sort
on training_case_tasks (case_id, sort_order, id);
create index if not exists idx_training_case_tasks_skill_type
on training_case_tasks (skill_tag, task_type);
create table if not exists training_attempts (
id bigserial primary key,
case_id bigint not null references training_cases(id),
user_id text not null default 'local',
status text not null default 'started',
current_frame_index integer not null default 0,
started_at timestamptz not null default now(),
submitted_at timestamptz,
total_score numeric(8,2),
meta jsonb not null default '{}'::jsonb,
created_at timestamptz not null default now(),
updated_at timestamptz not null default now()
);
create index if not exists idx_training_attempts_user_status_created_desc
on training_attempts (user_id, status, created_at desc);
create index if not exists idx_training_attempts_case_user_created_desc
on training_attempts (case_id, user_id, created_at desc);
create table if not exists training_annotations (
id bigserial primary key,
attempt_id bigint not null references training_attempts(id) on delete cascade,
task_id bigint references training_case_tasks(id),
user_id text not null default 'local',
annotation_type text not null,
payload jsonb not null default '{}'::jsonb,
created_at timestamptz not null default now(),
updated_at timestamptz not null default now()
);
create index if not exists idx_training_annotations_attempt_type_created
on training_annotations (attempt_id, annotation_type, created_at);
create table if not exists training_scores (
id bigserial primary key,
attempt_id bigint not null references training_attempts(id) on delete cascade,
task_id bigint references training_case_tasks(id),
user_id text not null default 'local',
category text not null,
score numeric(8,2) not null,
max_score numeric(8,2) not null,
feedback jsonb not null default '{}'::jsonb,
created_at timestamptz not null default now(),
updated_at timestamptz not null default now()
);
create index if not exists idx_training_scores_attempt_category
on training_scores (attempt_id, category);
create table if not exists training_progress (
id bigserial primary key,
user_id text not null default 'local',
skill_tag text not null,
attempt_count integer not null default 0,
avg_score numeric(8,4),
last_attempt_at timestamptz,
weakness_tags jsonb not null default '[]'::jsonb,
meta jsonb not null default '{}'::jsonb,
created_at timestamptz not null default now(),
updated_at timestamptz not null default now(),
unique (user_id, skill_tag)
);
create index if not exists idx_training_progress_user_skill
on training_progress (user_id, skill_tag);
+4 -1
View File
@@ -1,5 +1,8 @@
from sqlalchemy import DateTime
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
pass
type_annotation_map = {
object: DateTime(timezone=True),
}
+14
View File
@@ -15,6 +15,14 @@ from src.models.sweep_event import SweepEvent
from src.models.swing import Swing
from src.models.trade_execution import TradeExecution
from src.models.trade_signal import TradeSignal
from src.models.training import (
TrainingAnnotation,
TrainingAttempt,
TrainingCase,
TrainingCaseTask,
TrainingProgress,
TrainingScore,
)
from src.models.venue import Venue
__all__ = [
@@ -33,5 +41,11 @@ __all__ = [
"Swing",
"TradeExecution",
"TradeSignal",
"TrainingAnnotation",
"TrainingAttempt",
"TrainingCase",
"TrainingCaseTask",
"TrainingProgress",
"TrainingScore",
"Venue",
]
+108
View File
@@ -0,0 +1,108 @@
from datetime import datetime
from typing import Optional
from sqlalchemy import DateTime, ForeignKey, Integer, Numeric, Text, UniqueConstraint
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
from src.db.base import Base
class TrainingCase(Base):
__tablename__ = "training_cases"
id: Mapped[int] = mapped_column(primary_key=True)
instrument_id: Mapped[int] = mapped_column(ForeignKey("instruments.id"), nullable=False)
timeframe: Mapped[str] = mapped_column(Text, nullable=False)
case_type: Mapped[str] = mapped_column(Text, nullable=False)
skill_tag: Mapped[str] = mapped_column(Text, nullable=False)
chapter_ref: Mapped[Optional[str]] = mapped_column(Text)
source_event_type: Mapped[Optional[str]] = mapped_column(Text)
source_event_id: Mapped[Optional[int]] = mapped_column()
source_event_refs: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
start_ts: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
decision_ts: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
end_ts: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
difficulty: Mapped[str] = mapped_column(Text, nullable=False, default="medium")
answer_key: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
status: Mapped[str] = mapped_column(Text, nullable=False, default="draft")
meta: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
class TrainingCaseTask(Base):
__tablename__ = "training_case_tasks"
id: Mapped[int] = mapped_column(primary_key=True)
case_id: Mapped[int] = mapped_column(ForeignKey("training_cases.id"), nullable=False)
task_type: Mapped[str] = mapped_column(Text, nullable=False)
skill_tag: Mapped[str] = mapped_column(Text, nullable=False)
prompt: Mapped[str] = mapped_column(Text, nullable=False)
required_annotation_types: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
max_score: Mapped[float] = mapped_column(Numeric(8, 2), nullable=False, default=0)
sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
status: Mapped[str] = mapped_column(Text, nullable=False, default="active")
meta: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
class TrainingAttempt(Base):
__tablename__ = "training_attempts"
id: Mapped[int] = mapped_column(primary_key=True)
case_id: Mapped[int] = mapped_column(ForeignKey("training_cases.id"), nullable=False)
user_id: Mapped[str] = mapped_column(Text, nullable=False, default="local")
status: Mapped[str] = mapped_column(Text, nullable=False, default="started")
current_frame_index: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
submitted_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True))
total_score: Mapped[Optional[float]] = mapped_column(Numeric(8, 2))
meta: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
class TrainingAnnotation(Base):
__tablename__ = "training_annotations"
id: Mapped[int] = mapped_column(primary_key=True)
attempt_id: Mapped[int] = mapped_column(ForeignKey("training_attempts.id"), nullable=False)
task_id: Mapped[Optional[int]] = mapped_column(ForeignKey("training_case_tasks.id"))
user_id: Mapped[str] = mapped_column(Text, nullable=False, default="local")
annotation_type: Mapped[str] = mapped_column(Text, nullable=False)
payload: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
class TrainingScore(Base):
__tablename__ = "training_scores"
id: Mapped[int] = mapped_column(primary_key=True)
attempt_id: Mapped[int] = mapped_column(ForeignKey("training_attempts.id"), nullable=False)
task_id: Mapped[Optional[int]] = mapped_column(ForeignKey("training_case_tasks.id"))
user_id: Mapped[str] = mapped_column(Text, nullable=False, default="local")
category: Mapped[str] = mapped_column(Text, nullable=False)
score: Mapped[float] = mapped_column(Numeric(8, 2), nullable=False)
max_score: Mapped[float] = mapped_column(Numeric(8, 2), nullable=False)
feedback: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
class TrainingProgress(Base):
__tablename__ = "training_progress"
__table_args__ = (UniqueConstraint("user_id", "skill_tag", name="uq_training_progress_user_skill"),)
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[str] = mapped_column(Text, nullable=False, default="local")
skill_tag: Mapped[str] = mapped_column(Text, nullable=False)
attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
avg_score: Mapped[Optional[float]] = mapped_column(Numeric(8, 4))
last_attempt_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True))
weakness_tags: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
meta: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+2
View File
@@ -13,6 +13,7 @@ from src.repositories.signal_repository import SignalRepository
from src.repositories.structure_repository import StructureRepository
from src.repositories.swing_repository import SwingRepository
from src.repositories.sweep_repository import SweepRepository
from src.repositories.training_case_repository import TrainingCaseRepository
__all__ = [
"BacktestRepository",
@@ -28,4 +29,5 @@ __all__ = [
"StructureRepository",
"SwingRepository",
"SweepRepository",
"TrainingCaseRepository",
]
@@ -0,0 +1,428 @@
from datetime import datetime, timezone
import json
from typing import Optional
from sqlalchemy import text
from src.db.session import SessionLocal
class TrainingCaseRepository:
def create_case(
self,
instrument_id: int,
timeframe: str,
case_type: str,
skill_tag: str,
start_ts,
decision_ts,
end_ts,
source_event_refs: list,
answer_key: dict,
status: str = "draft",
chapter_ref: Optional[str] = None,
source_event_type: Optional[str] = None,
source_event_id: Optional[int] = None,
difficulty: str = "medium",
meta: Optional[dict] = None,
) -> int:
sql = text(
"""
insert into training_cases (
instrument_id,
timeframe,
case_type,
skill_tag,
chapter_ref,
source_event_type,
source_event_id,
source_event_refs,
start_ts,
decision_ts,
end_ts,
difficulty,
answer_key,
status,
meta
) values (
:instrument_id,
:timeframe,
:case_type,
:skill_tag,
:chapter_ref,
:source_event_type,
:source_event_id,
cast(:source_event_refs as jsonb),
:start_ts,
:decision_ts,
:end_ts,
:difficulty,
cast(:answer_key as jsonb),
:status,
cast(:meta as jsonb)
)
returning id
"""
)
with SessionLocal.begin() as session:
result = session.execute(
sql,
{
"instrument_id": instrument_id,
"timeframe": timeframe,
"case_type": case_type,
"skill_tag": skill_tag,
"chapter_ref": chapter_ref,
"source_event_type": source_event_type,
"source_event_id": source_event_id,
"source_event_refs": json.dumps(source_event_refs, default=str),
"start_ts": start_ts,
"decision_ts": decision_ts,
"end_ts": end_ts,
"difficulty": difficulty,
"answer_key": json.dumps(answer_key, default=str),
"status": status,
"meta": json.dumps(meta or {}, default=str),
},
)
return int(result.scalar_one())
def add_task(
self,
case_id: int,
task_type: str,
skill_tag: str,
prompt: str,
required_annotation_types: Optional[list] = None,
max_score: float = 0,
sort_order: int = 0,
status: str = "active",
meta: Optional[dict] = None,
) -> int:
sql = text(
"""
insert into training_case_tasks (
case_id,
task_type,
skill_tag,
prompt,
required_annotation_types,
max_score,
sort_order,
status,
meta
) values (
:case_id,
:task_type,
:skill_tag,
:prompt,
cast(:required_annotation_types as jsonb),
:max_score,
:sort_order,
:status,
cast(:meta as jsonb)
)
returning id
"""
)
with SessionLocal.begin() as session:
result = session.execute(
sql,
{
"case_id": case_id,
"task_type": task_type,
"skill_tag": skill_tag,
"prompt": prompt,
"required_annotation_types": json.dumps(required_annotation_types or [], default=str),
"max_score": max_score,
"sort_order": sort_order,
"status": status,
"meta": json.dumps(meta or {}, default=str),
},
)
return int(result.scalar_one())
def list_cases(
self,
status: Optional[str] = None,
skill_tag: Optional[str] = None,
instrument_id: Optional[int] = None,
limit: int = 100,
) -> list[dict]:
filters = []
params = {"limit": limit}
if status is not None:
filters.append("status = :status")
params["status"] = status
if skill_tag is not None:
filters.append("skill_tag = :skill_tag")
params["skill_tag"] = skill_tag
if instrument_id is not None:
filters.append("instrument_id = :instrument_id")
params["instrument_id"] = instrument_id
where_sql = f"where {' and '.join(filters)}" if filters else ""
sql = text(
f"""
select
id,
instrument_id,
timeframe,
case_type,
skill_tag,
chapter_ref,
source_event_type,
source_event_id,
source_event_refs,
start_ts,
decision_ts,
end_ts,
difficulty,
status,
meta,
created_at,
updated_at
from training_cases
{where_sql}
order by created_at desc
limit :limit
"""
)
with SessionLocal() as session:
rows = session.execute(sql, params).mappings().all()
return [dict(row) for row in rows]
def get_case(self, case_id: int, include_answer_key: bool = False) -> Optional[dict]:
answer_key_select = ", answer_key" if include_answer_key else ""
sql = text(
f"""
select
id,
instrument_id,
timeframe,
case_type,
skill_tag,
chapter_ref,
source_event_type,
source_event_id,
source_event_refs,
start_ts,
decision_ts,
end_ts,
difficulty,
status,
meta,
created_at,
updated_at
{answer_key_select}
from training_cases
where id = :case_id
limit 1
"""
)
with SessionLocal() as session:
row = session.execute(sql, {"case_id": case_id}).mappings().first()
return dict(row) if row else None
def create_attempt(
self,
case_id: int,
user_id: str = "local",
status: str = "started",
current_frame_index: int = 0,
meta: Optional[dict] = None,
) -> int:
sql = text(
"""
insert into training_attempts (
case_id,
user_id,
status,
current_frame_index,
started_at,
meta
) values (
:case_id,
:user_id,
:status,
:current_frame_index,
:started_at,
cast(:meta as jsonb)
)
returning id
"""
)
with SessionLocal.begin() as session:
result = session.execute(
sql,
{
"case_id": case_id,
"user_id": user_id,
"status": status,
"current_frame_index": current_frame_index,
"started_at": datetime.now(timezone.utc),
"meta": json.dumps(meta or {}, default=str),
},
)
return int(result.scalar_one())
def add_annotation(
self,
attempt_id: int,
annotation_type: str,
payload: dict,
user_id: str = "local",
task_id: Optional[int] = None,
) -> int:
sql = text(
"""
insert into training_annotations (
attempt_id,
task_id,
user_id,
annotation_type,
payload
) values (
:attempt_id,
:task_id,
:user_id,
:annotation_type,
cast(:payload as jsonb)
)
returning id
"""
)
with SessionLocal.begin() as session:
result = session.execute(
sql,
{
"attempt_id": attempt_id,
"task_id": task_id,
"user_id": user_id,
"annotation_type": annotation_type,
"payload": json.dumps(payload, default=str),
},
)
return int(result.scalar_one())
def submit_attempt(self, attempt_id: int, total_score: Optional[float], status: str = "submitted") -> None:
sql = text(
"""
update training_attempts
set status = :status,
submitted_at = :submitted_at,
total_score = :total_score,
updated_at = now()
where id = :attempt_id
"""
)
with SessionLocal.begin() as session:
session.execute(
sql,
{
"attempt_id": attempt_id,
"status": status,
"submitted_at": datetime.now(timezone.utc),
"total_score": total_score,
},
)
def submit_score(
self,
attempt_id: int,
category: str,
score: float,
max_score: float,
feedback: dict,
user_id: str = "local",
task_id: Optional[int] = None,
) -> int:
sql = text(
"""
insert into training_scores (
attempt_id,
task_id,
user_id,
category,
score,
max_score,
feedback
) values (
:attempt_id,
:task_id,
:user_id,
:category,
:score,
:max_score,
cast(:feedback as jsonb)
)
returning id
"""
)
with SessionLocal.begin() as session:
result = session.execute(
sql,
{
"attempt_id": attempt_id,
"task_id": task_id,
"user_id": user_id,
"category": category,
"score": score,
"max_score": max_score,
"feedback": json.dumps(feedback, default=str),
},
)
return int(result.scalar_one())
def update_progress(
self,
user_id: str,
skill_tag: str,
score: Optional[float],
weakness_tags: Optional[list] = None,
meta: Optional[dict] = None,
) -> None:
sql = text(
"""
insert into training_progress (
user_id,
skill_tag,
attempt_count,
avg_score,
last_attempt_at,
weakness_tags,
meta
) values (
:user_id,
:skill_tag,
1,
:score,
:last_attempt_at,
cast(:weakness_tags as jsonb),
cast(:meta as jsonb)
)
on conflict (user_id, skill_tag) do update
set attempt_count = training_progress.attempt_count + 1,
avg_score = case
when :score is null then training_progress.avg_score
when training_progress.avg_score is null then :score
else ((training_progress.avg_score * training_progress.attempt_count) + :score)
/ (training_progress.attempt_count + 1)
end,
last_attempt_at = excluded.last_attempt_at,
weakness_tags = excluded.weakness_tags,
meta = training_progress.meta || excluded.meta,
updated_at = now()
"""
)
with SessionLocal.begin() as session:
session.execute(
sql,
{
"user_id": user_id,
"skill_tag": skill_tag,
"score": score,
"last_attempt_at": datetime.now(timezone.utc),
"weakness_tags": json.dumps(weakness_tags or [], default=str),
"meta": json.dumps(meta or {}, default=str),
},
)
+166
View File
@@ -0,0 +1,166 @@
import inspect
import os
import re
import unittest
from unittest.mock import patch
os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")
from scripts import run_migration_static_checks
from src.db.base import Base
TRAINING_REQUIRED_SCHEMA = {
"training_cases": (
"id",
"case_type",
"skill_tag",
"instrument_id",
"timeframe",
"start_ts",
"decision_ts",
"end_ts",
"source_event_refs",
"answer_key",
"status",
"created_at",
"updated_at",
),
"training_case_tasks": (
"id",
"case_id",
"task_type",
"skill_tag",
"prompt",
"required_annotation_types",
"max_score",
"status",
"created_at",
"updated_at",
),
"training_attempts": (
"id",
"case_id",
"user_id",
"status",
"started_at",
"submitted_at",
"total_score",
"created_at",
"updated_at",
),
"training_annotations": (
"id",
"attempt_id",
"user_id",
"annotation_type",
"payload",
"created_at",
"updated_at",
),
"training_scores": (
"id",
"attempt_id",
"user_id",
"category",
"score",
"max_score",
"feedback",
"created_at",
"updated_at",
),
"training_progress": (
"id",
"user_id",
"skill_tag",
"attempt_count",
"avg_score",
"last_attempt_at",
"weakness_tags",
"created_at",
"updated_at",
),
}
class _Result:
def __init__(self, rows=None, scalar=1):
self._rows = rows or []
self._scalar = scalar
def mappings(self):
return self
def all(self):
return self._rows
def first(self):
return self._rows[0] if self._rows else None
def scalar_one(self):
return self._scalar
class _ReadSession:
def __enter__(self):
return self
def __exit__(self, exc_type, exc, traceback):
return False
def execute(self, sql, params=None):
sql_text = str(sql).lower()
case_row = {
"id": 7,
"case_type": "sweep",
"skill_tag": "liquidity_pool",
"instrument_id": 1,
"timeframe": "15m",
"start_ts": "2026-04-01T00:00:00Z",
"decision_ts": "2026-04-01T02:00:00Z",
"end_ts": "2026-04-01T04:00:00Z",
"source_event_refs": [{"table": "sweep_events", "id": 5}],
"status": "published",
"created_at": "2026-04-01T00:00:00Z",
"updated_at": "2026-04-01T00:00:00Z",
}
if "answer_key" in sql_text:
case_row["answer_key"] = {"side": "buy_side"}
return _Result(rows=[case_row])
class TrainingCaseEngineM1Tests(unittest.TestCase):
def test_migration_static_checks_can_identify_training_tables(self) -> None:
result = run_migration_static_checks.run_migration_static_checks(required_schema=TRAINING_REQUIRED_SCHEMA)
self.assertTrue(result["passed"], result["errors"])
def test_model_metadata_contains_training_tables(self) -> None:
import src.models # noqa: F401
for table_name in TRAINING_REQUIRED_SCHEMA:
self.assertIn(table_name, Base.metadata.tables)
def test_repository_source_does_not_write_to_trading_chain_tables(self) -> None:
from src.repositories.training_case_repository import TrainingCaseRepository
source = inspect.getsource(TrainingCaseRepository).lower()
for table_name in ("trade_signals", "trade_executions", "backtest_results"):
forbidden = re.compile(rf"\b(insert\s+into|update|delete\s+from|truncate\s+table)\s+{table_name}\b")
self.assertIsNone(forbidden.search(source), f"repository writes to {table_name}")
def test_list_cases_does_not_expose_answer_key_by_default(self) -> None:
from src.repositories import training_case_repository as repository_module
repository = repository_module.TrainingCaseRepository()
with patch.object(repository_module, "SessionLocal", return_value=_ReadSession()):
cases = repository.list_cases()
self.assertEqual(len(cases), 1)
self.assertNotIn("answer_key", cases[0])
if __name__ == "__main__":
unittest.main()