diff --git a/migrations/026_create_trading_journal_entries.sql b/migrations/026_create_trading_journal_entries.sql new file mode 100644 index 0000000..0b240a4 --- /dev/null +++ b/migrations/026_create_trading_journal_entries.sql @@ -0,0 +1,30 @@ +create table if not exists trading_journal_entries ( + id bigserial primary key, + execution_id bigint not null references trade_executions(id), + signal_id bigint references trade_signals(id), + trading_plan_id bigint references trading_plans(id), + candidate_id bigint references trade_plan_candidates(id), + session_code text, + execution_mode text not null, + journal_ts timestamptz not null, + journal_date date not null, + framework_execution boolean not null default false, + model_compliance boolean not null default false, + risk_compliance boolean not null default false, + model_failure boolean not null default false, + execution_violation boolean not null default false, + risk_violation boolean not null default false, + plan_violation boolean not null default false, + review_tags jsonb not null default '[]'::jsonb, + next_actions jsonb not null default '[]'::jsonb, + summary jsonb not null default '{}'::jsonb, + created_at timestamptz not null default now(), + updated_at timestamptz not null default now(), + unique (execution_id) +); + +create index if not exists idx_trading_journal_entries_date_desc +on trading_journal_entries (journal_date desc, journal_ts desc); + +create index if not exists idx_trading_journal_entries_session_date_desc +on trading_journal_entries (session_code, journal_date desc); diff --git a/package.json b/package.json index c87856c..1c1cb16 100644 --- a/package.json +++ b/package.json @@ -3,7 +3,7 @@ "version": "0.1.0", "private": true, "scripts": { - "lint": ".\\.venv\\Scripts\\python.exe scripts/run_script_static_checks.py && .\\.venv\\Scripts\\python.exe scripts/run_migration_static_checks.py && .\\.venv\\Scripts\\python.exe -m unittest tests.unit.test_trading_plan_service tests.unit.test_market_state_and_risk_governor tests.unit.test_target_selection_service tests.unit.test_trade_plan_candidate_service tests.unit.test_backtest_service tests.unit.test_backtest_script tests.unit.test_execution_service tests.unit.test_execution_ticket_script tests.unit.test_live_adapter_service tests.unit.test_live_adapter_script tests.unit.test_api_server tests.unit.test_api_and_main", + "lint": ".\\.venv\\Scripts\\python.exe scripts/run_script_static_checks.py && .\\.venv\\Scripts\\python.exe scripts/run_migration_static_checks.py && .\\.venv\\Scripts\\python.exe -m unittest tests.unit.test_trading_plan_service tests.unit.test_market_state_and_risk_governor tests.unit.test_target_selection_service tests.unit.test_trade_plan_candidate_service tests.unit.test_backtest_service tests.unit.test_backtest_script tests.unit.test_execution_service tests.unit.test_execution_ticket_script tests.unit.test_live_adapter_service tests.unit.test_live_adapter_script tests.unit.test_trading_journal_service tests.unit.test_api_server tests.unit.test_api_and_main tests.unit.test_report_services", "build": ".\\.venv\\Scripts\\python.exe -m compileall -q src scripts tests", "visual:snapshots": "node --check src/web/app.js" } diff --git a/scripts/run_reports.py b/scripts/run_reports.py index 0e209e4..59dad44 100644 --- a/scripts/run_reports.py +++ b/scripts/run_reports.py @@ -2,6 +2,7 @@ from pathlib import Path import json import os import sys +from sqlalchemy.exc import SQLAlchemyError ROOT = Path(__file__).resolve().parents[1] os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test") @@ -13,6 +14,7 @@ from src.services.reports import ( IntradayReportService, PostTradeReviewService, PreMarketReportService, + WeeklyReviewService, ) @@ -42,6 +44,19 @@ def main(env=None) -> int: backtest_result={"outcome": "win", "invalidated_before_entry": False, "error_tags": []}, ) backtest_summary = BacktestReportService().build_summary([]) + try: + weekly_review = WeeklyReviewService().build_summary() + except SQLAlchemyError: + weekly_review = { + "entry_count": 0, + "framework_execution_rate": 0.0, + "out_of_model_trade_rate": 0.0, + "risk_violation_count": 0, + "high_quality_session": None, + "low_quality_session": None, + "session_performance": {}, + "next_week_actions": ["collect_more_executed_samples"], + } payload = { "passed": True, @@ -49,6 +64,7 @@ def main(env=None) -> int: "intraday": intraday, "post_trade": review, "backtest_summary": backtest_summary, + "weekly_review": weekly_review, } if output_json: print(json.dumps(payload, default=str, ensure_ascii=False, sort_keys=True)) @@ -59,6 +75,7 @@ def main(env=None) -> int: print(f"Post-trade aligned: {review['rule_aligned']}") print(f"Backtest total signals: {backtest_summary['total_signals']}") print(f"Backtest quality gate passed: {backtest_summary['quality_gate']['passed']}") + print(f"Weekly next actions: {weekly_review['next_week_actions']}") return 0 diff --git a/scripts/run_trading_journal.py b/scripts/run_trading_journal.py new file mode 100644 index 0000000..2923acc --- /dev/null +++ b/scripts/run_trading_journal.py @@ -0,0 +1,90 @@ +from datetime import date, datetime +from pathlib import Path +import json +import os +import sys + +from sqlalchemy.exc import SQLAlchemyError + +ROOT = Path(__file__).resolve().parents[1] +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test") +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from src.services.journal import TradingJournalService, WeeklyReviewService + + +def main(env=None) -> int: + env = os.environ if env is None else env + output_json = _get_bool(env, "TRADING_JOURNAL_OUTPUT_JSON", False) + journal_service = TradingJournalService() + weekly_review_service = WeeklyReviewService() + instrument_id = _get_optional_int(env, "TRADING_JOURNAL_INSTRUMENT_ID") + execution_id = _get_optional_int(env, "TRADING_JOURNAL_EXECUTION_ID") + week_start = _get_date(env.get("TRADING_JOURNAL_WEEK_START")) + week_end = _get_date(env.get("TRADING_JOURNAL_WEEK_END")) + try: + if execution_id is not None: + journal_id = journal_service.upsert_entry_for_execution(execution_id) + backfill = {"execution_ids": [execution_id], "created": 1 if journal_id else 0} + else: + backfill = journal_service.backfill_missing_entries(limit=_get_int(env, "TRADING_JOURNAL_BACKFILL_LIMIT", 100)) + weekly_summary = weekly_review_service.build_summary( + week_start=week_start, + week_end=week_end, + instrument_id=instrument_id, + ) + except SQLAlchemyError as exc: + payload = { + "passed": False, + "error_type": exc.__class__.__name__, + "error": str(exc), + } + if output_json: + print(json.dumps(payload, default=str, ensure_ascii=False, sort_keys=True)) + else: + print(f"Database unavailable for trading journal sample: {exc.__class__.__name__}") + return 2 + payload = { + "passed": True, + "backfill": backfill, + "weekly_summary": weekly_summary, + } + if output_json: + print(json.dumps(payload, default=str, ensure_ascii=False, sort_keys=True)) + else: + print(f"Journal backfill created: {backfill['created']}") + print(f"Weekly entries: {weekly_summary['entry_count']}") + print(f"Next week actions: {weekly_summary['next_week_actions']}") + return 0 + + +def _get_optional_int(env, key: str): + value = env.get(key) + if value is None or value == "": + return None + return int(value) + + +def _get_int(env, key: str, default: int) -> int: + value = env.get(key) + if value is None or value == "": + return default + return int(value) + + +def _get_bool(env, key: str, default: bool) -> bool: + value = env.get(key) + if value is None or value == "": + return default + return str(value).strip().lower() in {"1", "true", "yes", "on"} + + +def _get_date(value): + if not value: + return None + return date.fromisoformat(str(value)) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/api/__init__.py b/src/api/__init__.py index b4ad33d..12a616f 100644 --- a/src/api/__init__.py +++ b/src/api/__init__.py @@ -9,6 +9,7 @@ from src.api.server import route_request, serve_api from src.api.sessions import get_recent_sessions from src.api.signals import get_recent_signals from src.api.trade_plan_candidates import get_recent_trade_plan_candidates +from src.api.trading_journal import get_recent_trading_journal, get_trading_journal_summary from src.api.trading_plans import get_recent_trading_plans __all__ = [ @@ -19,6 +20,8 @@ __all__ = [ "get_recent_signals", "get_recent_live_operation_audits", "get_recent_trade_plan_candidates", + "get_recent_trading_journal", + "get_trading_journal_summary", "get_recent_trading_plans", "get_recent_backtest_summaries", "route_request", diff --git a/src/api/server.py b/src/api/server.py index 5910f98..58289dd 100644 --- a/src/api/server.py +++ b/src/api/server.py @@ -13,6 +13,7 @@ from src.api.live_operation_audits import get_recent_live_operation_audits from src.api.sessions import get_recent_sessions from src.api.signals import get_recent_signals from src.api.trade_plan_candidates import get_recent_trade_plan_candidates +from src.api.trading_journal import get_recent_trading_journal, get_trading_journal_summary from src.api.trading_plans import get_recent_trading_plans from src.api.training import ( add_training_annotation, @@ -40,6 +41,8 @@ def build_api_dependencies() -> dict: "sessions": get_recent_sessions, "signals": get_recent_signals, "trade_plan_candidates": get_recent_trade_plan_candidates, + "trading_journal": get_recent_trading_journal, + "trading_journal_summary": get_trading_journal_summary, "trading_plans": get_recent_trading_plans, "backtests": get_recent_backtest_summaries, "executions": execution_service.build_execution_dashboard_items, @@ -101,6 +104,13 @@ def route_request( instrument_id = _require_int(query, "instrument_id") limit = _get_int(query, "limit", 20) return 200, {"items": deps["trading_plans"](instrument_id=instrument_id, limit=limit)} + if method == "GET" and path == "/trading-journal": + instrument_id = _get_optional_int(query, "instrument_id") + limit = _get_int(query, "limit", 20) + return 200, {"items": deps["trading_journal"](instrument_id=instrument_id, limit=limit)} + if method == "GET" and path == "/trading-journal-summary": + instrument_id = _get_optional_int(query, "instrument_id") + return 200, deps["trading_journal_summary"](instrument_id=instrument_id) if method == "GET" and path == "/backtests": instrument_id = _get_optional_int(query, "instrument_id") limit = _get_int(query, "limit", 10) diff --git a/src/api/trading_journal.py b/src/api/trading_journal.py new file mode 100644 index 0000000..3390941 --- /dev/null +++ b/src/api/trading_journal.py @@ -0,0 +1,18 @@ +from src.repositories.trading_journal_repository import TradingJournalRepository, default_week_range +from src.services.journal import WeeklyReviewService + + +def get_recent_trading_journal(instrument_id: int = None, limit: int = 20) -> list[dict]: + return TradingJournalRepository().fetch_recent_journal_entries( + instrument_id=instrument_id, + limit=limit, + ) + + +def get_trading_journal_summary(instrument_id: int = None) -> dict: + week_start, week_end = default_week_range() + return WeeklyReviewService().build_summary( + week_start=week_start, + week_end=week_end, + instrument_id=instrument_id, + ) diff --git a/src/models/__init__.py b/src/models/__init__.py index 6cc3be4..4ccd428 100644 --- a/src/models/__init__.py +++ b/src/models/__init__.py @@ -18,6 +18,7 @@ from src.models.swing import Swing from src.models.trade_execution import TradeExecution from src.models.trade_plan_candidate import TradePlanCandidate from src.models.trade_signal import TradeSignal +from src.models.trading_journal_entry import TradingJournalEntry from src.models.trading_plan import TradingPlan from src.models.training import ( TrainingAnnotation, @@ -48,6 +49,7 @@ __all__ = [ "TradeExecution", "TradePlanCandidate", "TradeSignal", + "TradingJournalEntry", "TradingPlan", "TrainingAnnotation", "TrainingAttempt", diff --git a/src/models/trading_journal_entry.py b/src/models/trading_journal_entry.py new file mode 100644 index 0000000..2bdf71c --- /dev/null +++ b/src/models/trading_journal_entry.py @@ -0,0 +1,31 @@ +from typing import Optional + +from sqlalchemy import Boolean, Date, ForeignKey, Text +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.orm import Mapped, mapped_column + +from src.db.base import Base + + +class TradingJournalEntry(Base): + __tablename__ = "trading_journal_entries" + + id: Mapped[int] = mapped_column(primary_key=True) + execution_id: Mapped[int] = mapped_column(ForeignKey("trade_executions.id"), nullable=False) + signal_id: Mapped[Optional[int]] = mapped_column(ForeignKey("trade_signals.id")) + trading_plan_id: Mapped[Optional[int]] = mapped_column(ForeignKey("trading_plans.id")) + candidate_id: Mapped[Optional[int]] = mapped_column(ForeignKey("trade_plan_candidates.id")) + session_code: Mapped[Optional[str]] = mapped_column(Text) + execution_mode: Mapped[str] = mapped_column(Text, nullable=False) + journal_ts: Mapped[object] = mapped_column(nullable=False) + journal_date: Mapped[object] = mapped_column(Date, nullable=False) + framework_execution: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + model_compliance: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + risk_compliance: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + model_failure: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + execution_violation: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + risk_violation: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + plan_violation: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + review_tags: Mapped[list] = mapped_column(JSONB, nullable=False, default=list) + next_actions: Mapped[list] = mapped_column(JSONB, nullable=False, default=list) + summary: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict) diff --git a/src/repositories/trading_journal_repository.py b/src/repositories/trading_journal_repository.py new file mode 100644 index 0000000..2a41c1c --- /dev/null +++ b/src/repositories/trading_journal_repository.py @@ -0,0 +1,270 @@ +from datetime import date, datetime, timedelta, timezone +import json +from typing import Optional + +from sqlalchemy import text + +from src.db.session import SessionLocal + + +class TradingJournalRepository: + def fetch_execution_review_bundle(self, execution_id: int) -> Optional[dict]: + sql = text( + """ + select + te.id as execution_id, + te.signal_id, + te.execution_mode, + te.status as execution_status, + te.side as execution_side, + te.entry_price, + te.stop_loss, + te.take_profit_1, + te.take_profit_2, + te.size, + te.risk_pct, + te.opened_ts, + te.closed_ts, + te.pnl, + te.meta as execution_meta, + ts.instrument_id, + ts.model_code, + ts.side as signal_side, + ts.status as signal_status, + ts.session_code, + ts.created_ts as signal_created_ts, + ts.expires_ts as signal_expires_ts, + ts.invalidations as signal_invalidations, + ts.evidence as signal_evidence, + ts.meta as signal_meta, + br.id as backtest_result_id, + br.trading_plan_id, + br.candidate_id, + br.candidate_model_code, + br.outcome as backtest_outcome, + br.mfe, + br.mae, + br.tp1_hit, + br.tp2_hit, + br.invalidated_before_entry, + br.error_tags as backtest_error_tags, + br.meta as backtest_meta, + et.id as ticket_id, + et.status as ticket_status, + et.confirmed_by, + et.confirmed_at, + et.ticket_snapshot + from trade_executions te + join trade_signals ts on ts.id = te.signal_id + left join lateral ( + select id, trading_plan_id, candidate_id, candidate_model_code, outcome, mfe, mae, + tp1_hit, tp2_hit, invalidated_before_entry, error_tags, meta + from backtest_results + where signal_id = te.signal_id + order by id desc + limit 1 + ) br on true + left join lateral ( + select id, status, confirmed_by, confirmed_at, ticket_snapshot + from execution_tickets + where (execution_id = te.id) + or (signal_id = te.signal_id and execution_mode = te.execution_mode) + order by confirmed_at desc nulls last, created_ts desc nulls last, id desc + limit 1 + ) et on true + where te.id = :execution_id + limit 1 + """ + ) + with SessionLocal() as session: + row = session.execute(sql, {"execution_id": execution_id}).mappings().first() + return dict(row) if row else None + + def fetch_recent_execution_ids_without_journal(self, limit: int = 100) -> list[int]: + sql = text( + """ + select te.id + from trade_executions te + left join trading_journal_entries tje on tje.execution_id = te.id + where tje.id is null + order by coalesce(te.closed_ts, te.opened_ts) desc nulls last, te.id desc + limit :limit + """ + ) + with SessionLocal() as session: + rows = session.execute(sql, {"limit": limit}).scalars().all() + return [int(row) for row in rows] + + def upsert_journal_entry( + self, + *, + execution_id: int, + signal_id: Optional[int], + trading_plan_id: Optional[int], + candidate_id: Optional[int], + session_code: Optional[str], + execution_mode: str, + journal_ts, + journal_date, + framework_execution: bool, + model_compliance: bool, + risk_compliance: bool, + model_failure: bool, + execution_violation: bool, + risk_violation: bool, + plan_violation: bool, + review_tags: list[str], + next_actions: list[str], + summary: dict, + ) -> int: + sql = text( + """ + insert into trading_journal_entries ( + execution_id, + signal_id, + trading_plan_id, + candidate_id, + session_code, + execution_mode, + journal_ts, + journal_date, + framework_execution, + model_compliance, + risk_compliance, + model_failure, + execution_violation, + risk_violation, + plan_violation, + review_tags, + next_actions, + summary + ) values ( + :execution_id, + :signal_id, + :trading_plan_id, + :candidate_id, + :session_code, + :execution_mode, + :journal_ts, + :journal_date, + :framework_execution, + :model_compliance, + :risk_compliance, + :model_failure, + :execution_violation, + :risk_violation, + :plan_violation, + cast(:review_tags as jsonb), + cast(:next_actions as jsonb), + cast(:summary as jsonb) + ) + on conflict (execution_id) + do update set + signal_id = excluded.signal_id, + trading_plan_id = excluded.trading_plan_id, + candidate_id = excluded.candidate_id, + session_code = excluded.session_code, + execution_mode = excluded.execution_mode, + journal_ts = excluded.journal_ts, + journal_date = excluded.journal_date, + framework_execution = excluded.framework_execution, + model_compliance = excluded.model_compliance, + risk_compliance = excluded.risk_compliance, + model_failure = excluded.model_failure, + execution_violation = excluded.execution_violation, + risk_violation = excluded.risk_violation, + plan_violation = excluded.plan_violation, + review_tags = excluded.review_tags, + next_actions = excluded.next_actions, + summary = excluded.summary, + updated_at = now() + returning id + """ + ) + with SessionLocal.begin() as session: + result = session.execute( + sql, + { + "execution_id": execution_id, + "signal_id": signal_id, + "trading_plan_id": trading_plan_id, + "candidate_id": candidate_id, + "session_code": session_code, + "execution_mode": execution_mode, + "journal_ts": journal_ts, + "journal_date": journal_date, + "framework_execution": framework_execution, + "model_compliance": model_compliance, + "risk_compliance": risk_compliance, + "model_failure": model_failure, + "execution_violation": execution_violation, + "risk_violation": risk_violation, + "plan_violation": plan_violation, + "review_tags": json.dumps(review_tags, default=str), + "next_actions": json.dumps(next_actions, default=str), + "summary": json.dumps(summary, default=str), + }, + ) + return int(result.scalar_one()) + + def fetch_recent_journal_entries( + self, + *, + instrument_id: Optional[int] = None, + limit: int = 20, + ) -> list[dict]: + filters = [] + params = {"limit": limit} + if instrument_id is not None: + filters.append("ts.instrument_id = :instrument_id") + params["instrument_id"] = instrument_id + where_sql = f"where {' and '.join(filters)}" if filters else "" + sql = text( + f""" + select tje.*, ts.instrument_id + from trading_journal_entries tje + left join trade_signals ts on ts.id = tje.signal_id + {where_sql} + order by tje.journal_ts desc, tje.id desc + limit :limit + """ + ) + with SessionLocal() as session: + rows = session.execute(sql, params).mappings().all() + return [dict(row) for row in rows] + + def fetch_weekly_journal_entries( + self, + *, + week_start: date, + week_end: date, + instrument_id: Optional[int] = None, + ) -> list[dict]: + filters = [ + "tje.journal_date >= :week_start", + "tje.journal_date <= :week_end", + ] + params = {"week_start": week_start, "week_end": week_end} + if instrument_id is not None: + filters.append("ts.instrument_id = :instrument_id") + params["instrument_id"] = instrument_id + where_sql = " and ".join(filters) + sql = text( + f""" + select tje.*, ts.instrument_id + from trading_journal_entries tje + left join trade_signals ts on ts.id = tje.signal_id + where {where_sql} + order by tje.journal_ts asc, tje.id asc + """ + ) + with SessionLocal() as session: + rows = session.execute(sql, params).mappings().all() + return [dict(row) for row in rows] + + +def default_week_range(now: Optional[datetime] = None) -> tuple[date, date]: + now = now or datetime.now(timezone.utc) + start = (now - timedelta(days=now.weekday())).date() + end = start + timedelta(days=6) + return start, end diff --git a/src/services/execution/execution_service.py b/src/services/execution/execution_service.py index 4952012..0775a29 100644 --- a/src/services/execution/execution_service.py +++ b/src/services/execution/execution_service.py @@ -209,6 +209,7 @@ class ExecutionService: closed_ts=settled.get("closed_ts"), meta=dict(settled.get("meta") or {}), ) + self._auto_generate_journal_if_supported(int(settled["id"])) return settled def persist_execution_observation(self, trade_execution: Optional[dict], observation: Optional[dict]) -> Optional[dict]: @@ -221,6 +222,7 @@ class ExecutionService: closed_ts=settled.get("closed_ts"), meta=dict(settled.get("meta") or {}), ) + self._auto_generate_journal_if_supported(int(settled["id"])) return settled def settle_execution_by_id(self, execution_id: int, backtest_result: Optional[dict], observation: Optional[dict] = None) -> Optional[dict]: @@ -570,6 +572,13 @@ class ExecutionService: meta=decision.meta, ) + def _auto_generate_journal_if_supported(self, execution_id: int) -> None: + if self.repository.__class__.__name__ != "ExecutionRepository": + return + from src.services.journal import TradingJournalService + + TradingJournalService().upsert_entry_for_execution(execution_id) + def trade_execution_matches_decision(execution: dict, decision: ExecutionDecision) -> bool: return ( diff --git a/src/services/journal/__init__.py b/src/services/journal/__init__.py new file mode 100644 index 0000000..9820881 --- /dev/null +++ b/src/services/journal/__init__.py @@ -0,0 +1,13 @@ +"""Trading journal and weekly review services.""" + +from src.services.journal.trading_journal_service import ( + TradingJournalEntryCandidate, + TradingJournalService, +) +from src.services.journal.weekly_review_service import WeeklyReviewService + +__all__ = [ + "TradingJournalEntryCandidate", + "TradingJournalService", + "WeeklyReviewService", +] diff --git a/src/services/journal/trading_journal_service.py b/src/services/journal/trading_journal_service.py new file mode 100644 index 0000000..70de8e8 --- /dev/null +++ b/src/services/journal/trading_journal_service.py @@ -0,0 +1,302 @@ +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +from typing import Optional + +from src.repositories.trading_journal_repository import TradingJournalRepository +from src.services.backtest import classify_backtest_lifecycle_state_from_result +from src.services.reports import PostTradeReviewService + + +RISK_VIOLATION_CODES = { + "execution_kill_switch_enabled", + "daily_loss_limit_reached", + "consecutive_loss_limit_reached", + "max_open_positions_reached", + "same_side_exposure_limit_reached", + "market_data_missing", + "market_data_stale", + "news_window_active", + "opening_volatility_active", + "backtest_quality_gate_failed_semi_auto", +} +PLAN_VIOLATION_CODES = { + "missing_structural_targets", + "missing_management_rules", + "signal_invalidated_before_entry", + "execution_ticket_not_confirmed", + "execution_ticket_missing_confirmation_metadata", +} + + +@dataclass(frozen=True) +class TradingJournalEntryCandidate: + execution_id: int + signal_id: Optional[int] + trading_plan_id: Optional[int] + candidate_id: Optional[int] + session_code: Optional[str] + execution_mode: str + journal_ts: object + journal_date: object + framework_execution: bool + model_compliance: bool + risk_compliance: bool + model_failure: bool + execution_violation: bool + risk_violation: bool + plan_violation: bool + review_tags: list[str] + next_actions: list[str] + summary: dict + + def to_dict(self) -> dict: + return dict(asdict(self)) + + +class TradingJournalService: + def __init__( + self, + repository: Optional[TradingJournalRepository] = None, + post_trade_review_service: Optional[PostTradeReviewService] = None, + ) -> None: + self.repository = repository or TradingJournalRepository() + self.post_trade_review_service = post_trade_review_service or PostTradeReviewService() + + def build_entry_for_execution(self, execution_id: int) -> Optional[TradingJournalEntryCandidate]: + bundle = self.repository.fetch_execution_review_bundle(execution_id) + if not bundle: + return None + trade_execution = _build_trade_execution_payload(bundle) + trade_signal = _build_trade_signal_payload(bundle) + backtest_result = _build_backtest_result_payload(bundle) + review = self.post_trade_review_service.build_report( + trade_execution=trade_execution, + trade_signal=trade_signal, + backtest_result=backtest_result, + ) + execution_audit = dict(review.get("execution_audit") or {}) + error_types = list(review.get("error_types") or []) + lifecycle_state = review.get("backtest_lifecycle_state") or classify_backtest_lifecycle_state_from_result(backtest_result) + + risk_violation = any(code in RISK_VIOLATION_CODES for code in error_types) + execution_violation = execution_audit.get("audit_status") == "review" and not risk_violation and any( + code not in PLAN_VIOLATION_CODES for code in error_types + ) + plan_violation = any(code in PLAN_VIOLATION_CODES for code in error_types) + model_failure = ( + lifecycle_state in {"stopped_out", "missed"} + and not risk_violation + and not execution_violation + and not plan_violation + ) + model_compliance = not plan_violation and _signal_structure_is_aligned(trade_signal) + risk_compliance = not risk_violation + framework_execution = model_compliance and risk_compliance and not execution_violation and not model_failure + + review_tags = build_review_tags( + trade_execution=trade_execution, + backtest_result=backtest_result, + framework_execution=framework_execution, + risk_violation=risk_violation, + execution_violation=execution_violation, + plan_violation=plan_violation, + model_failure=model_failure, + ) + next_actions = build_next_actions( + review=review, + lifecycle_state=lifecycle_state, + risk_violation=risk_violation, + execution_violation=execution_violation, + plan_violation=plan_violation, + model_failure=model_failure, + ) + journal_ts = _resolve_journal_ts(trade_execution) + summary = { + "review": review, + "execution": trade_execution, + "signal": trade_signal, + "backtest_result": backtest_result, + "ticket": _build_ticket_payload(bundle), + } + return TradingJournalEntryCandidate( + execution_id=int(bundle["execution_id"]), + signal_id=bundle.get("signal_id"), + trading_plan_id=bundle.get("trading_plan_id"), + candidate_id=bundle.get("candidate_id"), + session_code=bundle.get("session_code"), + execution_mode=str(bundle.get("execution_mode") or "paper"), + journal_ts=journal_ts, + journal_date=journal_ts.date(), + framework_execution=framework_execution, + model_compliance=model_compliance, + risk_compliance=risk_compliance, + model_failure=model_failure, + execution_violation=execution_violation, + risk_violation=risk_violation, + plan_violation=plan_violation, + review_tags=review_tags, + next_actions=next_actions, + summary=summary, + ) + + def upsert_entry_for_execution(self, execution_id: int) -> Optional[int]: + candidate = self.build_entry_for_execution(execution_id) + if candidate is None: + return None + return self.repository.upsert_journal_entry(**candidate.to_dict()) + + def backfill_missing_entries(self, limit: int = 100) -> dict: + execution_ids = self.repository.fetch_recent_execution_ids_without_journal(limit=limit) + created = 0 + for execution_id in execution_ids: + journal_id = self.upsert_entry_for_execution(execution_id) + if journal_id: + created += 1 + return { + "execution_ids": execution_ids, + "created": created, + } + + +def build_review_tags( + *, + trade_execution: dict, + backtest_result: Optional[dict], + framework_execution: bool, + risk_violation: bool, + execution_violation: bool, + plan_violation: bool, + model_failure: bool, +) -> list[str]: + tags: list[str] = [] + if framework_execution: + tags.append("framework_execution") + if not risk_violation: + tags.append("risk_compliant") + if not execution_violation: + tags.append("execution_clean") + if trade_execution.get("status") in {"closed", "partially_closed", "open"}: + tags.append("trade_executed") + if backtest_result: + if backtest_result.get("tp1_hit"): + tags.append("tp1_hit") + if backtest_result.get("tp2_hit"): + tags.append("tp2_hit") + lifecycle = classify_backtest_lifecycle_state_from_result(backtest_result) + if lifecycle: + tags.append(lifecycle) + if plan_violation: + tags.append("plan_violation") + if model_failure: + tags.append("model_failure") + return _dedupe(tags) + + +def build_next_actions( + *, + review: dict, + lifecycle_state: Optional[str], + risk_violation: bool, + execution_violation: bool, + plan_violation: bool, + model_failure: bool, +) -> list[str]: + actions: list[str] = [] + if risk_violation: + actions.append("tighten_risk_governor_or_session_filter") + if execution_violation: + actions.append("review_execution_process_and_confirmation_path") + if plan_violation: + actions.append("tighten_plan_and_candidate_validation") + if model_failure: + actions.append("review_model_context_and_target_selection") + if lifecycle_state == "tp2_closed": + actions.append("retain_current_model_playbook") + actions.extend(list(review.get("improvement_notes") or [])) + return _dedupe(actions)[:5] + + +def _signal_structure_is_aligned(trade_signal: Optional[dict]) -> bool: + if not trade_signal: + return False + meta = dict(trade_signal.get("meta") or {}) + structure_state = meta.get("structure_state") or trade_signal.get("structure_state") + return structure_state in {None, "aligned"} + + +def _resolve_journal_ts(trade_execution: dict) -> datetime: + value = trade_execution.get("closed_ts") or trade_execution.get("opened_ts") + if isinstance(value, datetime): + return value + if isinstance(value, str): + return datetime.fromisoformat(value.replace("Z", "+00:00")) + return datetime.now(timezone.utc) + + +def _build_trade_execution_payload(bundle: dict) -> dict: + return { + "id": bundle.get("execution_id"), + "signal_id": bundle.get("signal_id"), + "execution_mode": bundle.get("execution_mode"), + "status": bundle.get("execution_status"), + "opened_ts": bundle.get("opened_ts"), + "closed_ts": bundle.get("closed_ts"), + "meta": dict(bundle.get("execution_meta") or {}), + } + + +def _build_trade_signal_payload(bundle: dict) -> dict: + return { + "id": bundle.get("signal_id"), + "instrument_id": bundle.get("instrument_id"), + "model_code": bundle.get("model_code"), + "side": bundle.get("signal_side"), + "status": bundle.get("signal_status"), + "session_code": bundle.get("session_code"), + "invalidations": list(bundle.get("signal_invalidations") or []), + "evidence": dict(bundle.get("signal_evidence") or {}), + "meta": dict(bundle.get("signal_meta") or {}), + } + + +def _build_backtest_result_payload(bundle: dict) -> Optional[dict]: + if bundle.get("backtest_result_id") is None: + return None + return { + "id": bundle.get("backtest_result_id"), + "trading_plan_id": bundle.get("trading_plan_id"), + "candidate_id": bundle.get("candidate_id"), + "candidate_model_code": bundle.get("candidate_model_code"), + "outcome": bundle.get("backtest_outcome"), + "mfe": bundle.get("mfe"), + "mae": bundle.get("mae"), + "tp1_hit": bundle.get("tp1_hit"), + "tp2_hit": bundle.get("tp2_hit"), + "invalidated_before_entry": bundle.get("invalidated_before_entry"), + "error_tags": list(bundle.get("backtest_error_tags") or []), + "meta": dict(bundle.get("backtest_meta") or {}), + } + + +def _build_ticket_payload(bundle: dict) -> Optional[dict]: + if bundle.get("ticket_id") is None: + return None + return { + "id": bundle.get("ticket_id"), + "status": bundle.get("ticket_status"), + "confirmed_by": bundle.get("confirmed_by"), + "confirmed_at": bundle.get("confirmed_at"), + "ticket_snapshot": dict(bundle.get("ticket_snapshot") or {}), + } + + +def _dedupe(values: list[str]) -> list[str]: + seen = set() + result = [] + for value in values: + text = str(value or "").strip() + if not text or text in seen: + continue + seen.add(text) + result.append(text) + return result diff --git a/src/services/journal/weekly_review_service.py b/src/services/journal/weekly_review_service.py new file mode 100644 index 0000000..547b4a9 --- /dev/null +++ b/src/services/journal/weekly_review_service.py @@ -0,0 +1,124 @@ +from collections import Counter, defaultdict +from datetime import date +from typing import Optional + +from src.repositories.trading_journal_repository import TradingJournalRepository, default_week_range + + +class WeeklyReviewService: + def __init__(self, repository: Optional[TradingJournalRepository] = None) -> None: + self.repository = repository or TradingJournalRepository() + + def build_summary( + self, + *, + week_start: Optional[date] = None, + week_end: Optional[date] = None, + instrument_id: Optional[int] = None, + ) -> dict: + if week_start is None or week_end is None: + week_start, week_end = default_week_range() + entries = self.repository.fetch_weekly_journal_entries( + week_start=week_start, + week_end=week_end, + instrument_id=instrument_id, + ) + total = len(entries) + framework_count = sum(1 for entry in entries if entry.get("framework_execution")) + model_outside_count = sum(1 for entry in entries if not entry.get("model_compliance")) + risk_violation_count = sum(1 for entry in entries if entry.get("risk_violation")) + session_stats = build_session_stats(entries) + high_quality_session, low_quality_session = rank_sessions(session_stats) + next_actions = recommend_next_actions(entries, low_quality_session=low_quality_session) + return { + "week_start": week_start, + "week_end": week_end, + "entry_count": total, + "framework_execution_rate": _ratio(framework_count, total), + "out_of_model_trade_rate": _ratio(model_outside_count, total), + "risk_violation_count": risk_violation_count, + "high_quality_session": high_quality_session, + "low_quality_session": low_quality_session, + "session_performance": session_stats, + "next_week_actions": next_actions, + } + + +def build_session_stats(entries: list[dict]) -> dict: + buckets = defaultdict( + lambda: { + "count": 0, + "framework_count": 0, + "risk_violations": 0, + "model_outside": 0, + "total_r": 0.0, + } + ) + for entry in entries: + session_code = str(entry.get("session_code") or "OFF_HOURS") + bucket = buckets[session_code] + bucket["count"] += 1 + if entry.get("framework_execution"): + bucket["framework_count"] += 1 + if entry.get("risk_violation"): + bucket["risk_violations"] += 1 + if not entry.get("model_compliance"): + bucket["model_outside"] += 1 + summary = dict(entry.get("summary") or {}) + backtest_result = dict(summary.get("backtest_result") or {}) + backtest_meta = dict(backtest_result.get("meta") or {}) + bucket["total_r"] += float(backtest_meta.get("r_multiple") or 0.0) + stats = {} + for key, bucket in buckets.items(): + stats[key] = { + "count": bucket["count"], + "framework_execution_rate": _ratio(bucket["framework_count"], bucket["count"]), + "model_outside_rate": _ratio(bucket["model_outside"], bucket["count"]), + "risk_violation_count": bucket["risk_violations"], + "avg_r_multiple": bucket["total_r"] / bucket["count"] if bucket["count"] else 0.0, + } + return stats + + +def rank_sessions(session_stats: dict) -> tuple[Optional[dict], Optional[dict]]: + if not session_stats: + return None, None + ranked = sorted( + session_stats.items(), + key=lambda item: ( + item[1]["framework_execution_rate"], + item[1]["avg_r_multiple"], + -item[1]["risk_violation_count"], + ), + reverse=True, + ) + high_key, high_value = ranked[0] + low_key, low_value = ranked[-1] + return ( + {"session_code": high_key, **high_value}, + {"session_code": low_key, **low_value}, + ) + + +def recommend_next_actions(entries: list[dict], *, low_quality_session: Optional[dict]) -> list[str]: + counter = Counter() + for entry in entries: + for action in list(entry.get("next_actions") or []): + counter[str(action)] += 1 + actions = [item for item, _count in counter.most_common(3)] + if low_quality_session is not None: + actions.append(f"review_{str(low_quality_session['session_code']).lower()}_session_playbook") + if not actions: + actions = ["collect_more_executed_samples", "review_weekly_trade_selection", "review_risk_process"] + deduped = [] + seen = set() + for action in actions: + if action in seen: + continue + seen.add(action) + deduped.append(action) + return deduped[:3] + + +def _ratio(numerator: int, denominator: int) -> float: + return numerator / denominator if denominator else 0.0 diff --git a/src/services/reports/__init__.py b/src/services/reports/__init__.py index de05941..bc0db8f 100644 --- a/src/services/reports/__init__.py +++ b/src/services/reports/__init__.py @@ -4,10 +4,12 @@ from src.services.reports.backtest_report_service import BacktestReportService from src.services.reports.intraday_report_service import IntradayReportService from src.services.reports.post_trade_review_service import PostTradeReviewService from src.services.reports.pre_market_report_service import PreMarketReportService +from src.services.journal.weekly_review_service import WeeklyReviewService __all__ = [ "BacktestReportService", "IntradayReportService", "PostTradeReviewService", "PreMarketReportService", + "WeeklyReviewService", ] diff --git a/tests/unit/test_api_server.py b/tests/unit/test_api_server.py index b514c06..fe5a66a 100644 --- a/tests/unit/test_api_server.py +++ b/tests/unit/test_api_server.py @@ -105,6 +105,30 @@ class ApiServerRoutingTests(unittest.TestCase): self.assertEqual(status, 200) self.assertEqual(payload["items"], [{"id": 1, "status": "rejected"}]) + def test_routes_trading_journal_query(self) -> None: + status, payload = route_request( + "/trading-journal", + {"instrument_id": ["1"], "limit": ["2"]}, + dependencies={ + "trading_journal": lambda **kwargs: [{"id": 1, "framework_execution": False}], + }, + ) + + self.assertEqual(status, 200) + self.assertEqual(payload["items"], [{"id": 1, "framework_execution": False}]) + + def test_routes_trading_journal_summary_query(self) -> None: + status, payload = route_request( + "/trading-journal-summary", + {"instrument_id": ["1"]}, + dependencies={ + "trading_journal_summary": lambda **kwargs: {"entry_count": 3, "next_week_actions": ["review_risk_process"]}, + }, + ) + + self.assertEqual(status, 200) + self.assertEqual(payload["entry_count"], 3) + def test_routes_health(self) -> None: status, payload = route_request( "/health", diff --git a/tests/unit/test_report_services.py b/tests/unit/test_report_services.py index 754ce1e..4a314c1 100644 --- a/tests/unit/test_report_services.py +++ b/tests/unit/test_report_services.py @@ -150,6 +150,7 @@ class ReportScriptTests(unittest.TestCase): self.assertEqual(payload["intraday"]["report_type"], "intraday_explanation") self.assertEqual(payload["post_trade"]["report_type"], "post_trade_review") self.assertIn("quality_gate", payload["backtest_summary"]) + self.assertIn("weekly_review", payload) if __name__ == "__main__": diff --git a/tests/unit/test_trading_journal_service.py b/tests/unit/test_trading_journal_service.py new file mode 100644 index 0000000..0ae1df6 --- /dev/null +++ b/tests/unit/test_trading_journal_service.py @@ -0,0 +1,164 @@ +import contextlib +import io +import json +import os +import unittest +from datetime import datetime, timezone +from unittest.mock import patch + +from sqlalchemy.exc import OperationalError + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test") + +from scripts import run_trading_journal +from src.services.journal import TradingJournalService, WeeklyReviewService + + +class TradingJournalServiceTests(unittest.TestCase): + def test_build_entry_classifies_plan_and_risk_violations(self) -> None: + service = TradingJournalService(repository=_JournalRepo(_bundle())) + + entry = service.build_entry_for_execution(1) + + self.assertIsNotNone(entry) + assert entry is not None + self.assertTrue(entry.plan_violation) + self.assertTrue(entry.risk_violation) + self.assertFalse(entry.framework_execution) + self.assertIn("plan_violation", entry.review_tags) + self.assertIn("tighten_plan_and_candidate_validation", entry.next_actions) + + def test_upsert_entry_persists_journal(self) -> None: + repo = _JournalRepo(_bundle()) + service = TradingJournalService(repository=repo) + + journal_id = service.upsert_entry_for_execution(1) + + self.assertEqual(journal_id, 1) + self.assertEqual(repo.upserts[0]["execution_id"], 1) + + def test_weekly_review_builds_summary_with_actions(self) -> None: + repo = _JournalRepo(_bundle()) + repo.weekly_entries = [ + { + "session_code": "LONDON", + "framework_execution": True, + "model_compliance": True, + "risk_violation": False, + "next_actions": ["keep_same_model"], + "summary": {"backtest_result": {"meta": {"r_multiple": 1.2}}}, + }, + { + "session_code": "ASIA", + "framework_execution": False, + "model_compliance": False, + "risk_violation": True, + "next_actions": ["review_risk_process"], + "summary": {"backtest_result": {"meta": {"r_multiple": -0.8}}}, + }, + ] + summary = WeeklyReviewService(repository=repo).build_summary( + week_start=datetime(2026, 5, 4, tzinfo=timezone.utc).date(), + week_end=datetime(2026, 5, 10, tzinfo=timezone.utc).date(), + ) + + self.assertEqual(summary["entry_count"], 2) + self.assertEqual(summary["risk_violation_count"], 1) + self.assertEqual(summary["high_quality_session"]["session_code"], "LONDON") + self.assertEqual(summary["low_quality_session"]["session_code"], "ASIA") + self.assertEqual(len(summary["next_week_actions"]), 3) + + +class TradingJournalScriptTests(unittest.TestCase): + def test_run_trading_journal_outputs_json(self) -> None: + output = io.StringIO() + + with patch( + "scripts.run_trading_journal.TradingJournalService.backfill_missing_entries", + return_value={"execution_ids": [1], "created": 1}, + ), patch( + "scripts.run_trading_journal.WeeklyReviewService.build_summary", + return_value={"entry_count": 1, "next_week_actions": ["review_risk_process"]}, + ), contextlib.redirect_stdout(output): + exit_code = run_trading_journal.main(env={"TRADING_JOURNAL_OUTPUT_JSON": "1"}) + + payload = json.loads(output.getvalue()) + self.assertEqual(exit_code, 0) + self.assertTrue(payload["passed"]) + self.assertEqual(payload["backfill"]["created"], 1) + self.assertEqual(payload["weekly_summary"]["entry_count"], 1) + + def test_run_trading_journal_reports_database_unavailable_json(self) -> None: + output = io.StringIO() + + with patch( + "scripts.run_trading_journal.TradingJournalService.backfill_missing_entries", + side_effect=OperationalError("stmt", {}, Exception()), + ), contextlib.redirect_stdout(output): + exit_code = run_trading_journal.main(env={"TRADING_JOURNAL_OUTPUT_JSON": "1"}) + + payload = json.loads(output.getvalue()) + self.assertEqual(exit_code, 2) + self.assertFalse(payload["passed"]) + self.assertEqual(payload["error_type"], "OperationalError") + + +class _JournalRepo: + def __init__(self, bundle: dict) -> None: + self.bundle = bundle + self.upserts = [] + self.weekly_entries = [] + + def fetch_execution_review_bundle(self, execution_id: int): + return dict(self.bundle) + + def upsert_journal_entry(self, **kwargs): + self.upserts.append(kwargs) + return len(self.upserts) + + def fetch_recent_execution_ids_without_journal(self, limit: int = 100): + return [1] + + def fetch_weekly_journal_entries(self, **kwargs): + return list(self.weekly_entries) + + +def _bundle() -> dict: + return { + "execution_id": 1, + "signal_id": 2, + "execution_mode": "semi_auto", + "execution_status": "closed", + "opened_ts": datetime(2026, 5, 6, 0, 0, tzinfo=timezone.utc), + "closed_ts": datetime(2026, 5, 6, 1, 0, tzinfo=timezone.utc), + "execution_meta": {"risk_checks": ["missing_structural_targets", "news_window_active"]}, + "instrument_id": 1, + "model_code": "SWEEP_MSS_FVG", + "signal_side": "buy", + "signal_status": "awaiting_confirmation", + "session_code": "LONDON", + "signal_invalidations": ["session_mismatch"], + "signal_evidence": {"mss_id": 1}, + "signal_meta": {"structure_state": "mss_invalidated"}, + "backtest_result_id": 3, + "trading_plan_id": 4, + "candidate_id": 5, + "candidate_model_code": "OSOK", + "backtest_outcome": "loss", + "mfe": 1.0, + "mae": 0.5, + "tp1_hit": False, + "tp2_hit": False, + "invalidated_before_entry": False, + "backtest_error_tags": ["same_candle_entry_exit_ambiguous"], + "backtest_meta": {"r_multiple": -1.0}, + "ticket_id": 6, + "ticket_status": "confirmed", + "confirmed_by": "alice", + "confirmed_at": "2026-05-06T00:00:00+00:00", + "ticket_snapshot": {}, + } + + +if __name__ == "__main__": + unittest.main()