Compare commits

..

3 Commits

Author SHA1 Message Date
agentson
2f9efdad64 feat: integrate decision logger with main trading loop
Some checks failed
CI / test (pull_request) Has been cancelled
- Add DecisionLogger to main.py trading cycle
- Log all decisions with context snapshot (L1-L2 layers)
- Capture market data and balance info in context
- Add comprehensive tests (9 tests, 100% coverage)
- All tests passing (63 total)

Implements issue #17 acceptance criteria:
-  decision_logs table with proper schema
-  DecisionLogger class with all required methods
-  Automatic logging in trading loop
-  Tests achieve 100% coverage of decision_logger.py
- ⚠️  Context snapshot uses L1-L2 data (L3-L7 pending issue #15)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-02-04 15:47:53 +09:00
agentson
6551d7af79 WIP: Add decision logging infrastructure
- Add decision_logs table to database schema
- Create decision logger module with comprehensive logging
- Prepare for decision tracking and audit trail

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-02-04 15:47:53 +09:00
7515a5a314 Merge pull request 'feat: implement L1-L7 context tree for multi-layered memory management' (#16) from feature/issue-15-context-tree into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #16
2026-02-04 15:40:00 +09:00
5 changed files with 601 additions and 0 deletions

View File

@@ -55,6 +55,28 @@ def init_db(db_path: str) -> sqlite3.Connection:
"""
)
# Decision logging table for comprehensive audit trail
conn.execute(
"""
CREATE TABLE IF NOT EXISTS decision_logs (
decision_id TEXT PRIMARY KEY,
timestamp TEXT NOT NULL,
stock_code TEXT NOT NULL,
market TEXT NOT NULL,
exchange_code TEXT NOT NULL,
action TEXT NOT NULL,
confidence INTEGER NOT NULL,
rationale TEXT NOT NULL,
context_snapshot TEXT NOT NULL,
input_data TEXT NOT NULL,
outcome_pnl REAL,
outcome_accuracy INTEGER,
reviewed INTEGER DEFAULT 0,
review_notes TEXT
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS context_metadata (
@@ -71,6 +93,16 @@ def init_db(db_path: str) -> sqlite3.Connection:
conn.execute("CREATE INDEX IF NOT EXISTS idx_contexts_timeframe ON contexts(timeframe)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_contexts_updated ON contexts(updated_at)")
# Create indices for efficient decision log queries
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_decision_logs_timestamp ON decision_logs(timestamp)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_decision_logs_reviewed ON decision_logs(reviewed)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_decision_logs_confidence ON decision_logs(confidence)"
)
conn.commit()
return conn

5
src/logging/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
"""Decision logging and audit trail for trade decisions."""
from src.logging.decision_logger import DecisionLog, DecisionLogger
__all__ = ["DecisionLog", "DecisionLogger"]

View File

@@ -0,0 +1,235 @@
"""Decision logging system with context snapshots for comprehensive audit trail."""
from __future__ import annotations
import json
import sqlite3
import uuid
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import Any
@dataclass
class DecisionLog:
"""A logged trading decision with context and outcome."""
decision_id: str
timestamp: str
stock_code: str
market: str
exchange_code: str
action: str
confidence: int
rationale: str
context_snapshot: dict[str, Any]
input_data: dict[str, Any]
outcome_pnl: float | None = None
outcome_accuracy: int | None = None
reviewed: bool = False
review_notes: str | None = None
class DecisionLogger:
"""Logs trading decisions with full context for review and evolution."""
def __init__(self, conn: sqlite3.Connection) -> None:
"""Initialize the decision logger with a database connection."""
self.conn = conn
def log_decision(
self,
stock_code: str,
market: str,
exchange_code: str,
action: str,
confidence: int,
rationale: str,
context_snapshot: dict[str, Any],
input_data: dict[str, Any],
) -> str:
"""Log a trading decision with full context.
Args:
stock_code: Stock symbol
market: Market code (e.g., "KR", "US_NASDAQ")
exchange_code: Exchange code (e.g., "KRX", "NASDAQ")
action: Trading action (BUY/SELL/HOLD)
confidence: Confidence level (0-100)
rationale: Reasoning for the decision
context_snapshot: L1-L7 context snapshot at decision time
input_data: Market data inputs (price, volume, orderbook, etc.)
Returns:
decision_id: Unique identifier for this decision
"""
decision_id = str(uuid.uuid4())
timestamp = datetime.now(UTC).isoformat()
self.conn.execute(
"""
INSERT INTO decision_logs (
decision_id, timestamp, stock_code, market, exchange_code,
action, confidence, rationale, context_snapshot, input_data
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
decision_id,
timestamp,
stock_code,
market,
exchange_code,
action,
confidence,
rationale,
json.dumps(context_snapshot),
json.dumps(input_data),
),
)
self.conn.commit()
return decision_id
def get_unreviewed_decisions(
self, min_confidence: int = 80, limit: int | None = None
) -> list[DecisionLog]:
"""Get unreviewed decisions with high confidence.
Args:
min_confidence: Minimum confidence threshold (default 80)
limit: Maximum number of results (None = unlimited)
Returns:
List of unreviewed DecisionLog objects
"""
query = """
SELECT
decision_id, timestamp, stock_code, market, exchange_code,
action, confidence, rationale, context_snapshot, input_data,
outcome_pnl, outcome_accuracy, reviewed, review_notes
FROM decision_logs
WHERE reviewed = 0 AND confidence >= ?
ORDER BY timestamp DESC
"""
if limit is not None:
query += f" LIMIT {limit}"
cursor = self.conn.execute(query, (min_confidence,))
return [self._row_to_decision_log(row) for row in cursor.fetchall()]
def mark_reviewed(self, decision_id: str, notes: str) -> None:
"""Mark a decision as reviewed with notes.
Args:
decision_id: Decision identifier
notes: Review notes and insights
"""
self.conn.execute(
"""
UPDATE decision_logs
SET reviewed = 1, review_notes = ?
WHERE decision_id = ?
""",
(notes, decision_id),
)
self.conn.commit()
def update_outcome(
self, decision_id: str, pnl: float, accuracy: int
) -> None:
"""Update the outcome of a decision after trade execution.
Args:
decision_id: Decision identifier
pnl: Actual profit/loss realized
accuracy: 1 if decision was correct, 0 if wrong
"""
self.conn.execute(
"""
UPDATE decision_logs
SET outcome_pnl = ?, outcome_accuracy = ?
WHERE decision_id = ?
""",
(pnl, accuracy, decision_id),
)
self.conn.commit()
def get_decision_by_id(self, decision_id: str) -> DecisionLog | None:
"""Get a specific decision by ID.
Args:
decision_id: Decision identifier
Returns:
DecisionLog object or None if not found
"""
cursor = self.conn.execute(
"""
SELECT
decision_id, timestamp, stock_code, market, exchange_code,
action, confidence, rationale, context_snapshot, input_data,
outcome_pnl, outcome_accuracy, reviewed, review_notes
FROM decision_logs
WHERE decision_id = ?
""",
(decision_id,),
)
row = cursor.fetchone()
return self._row_to_decision_log(row) if row else None
def get_losing_decisions(
self, min_confidence: int = 80, min_loss: float = -100.0
) -> list[DecisionLog]:
"""Get high-confidence decisions that resulted in losses.
Useful for identifying patterns in failed predictions.
Args:
min_confidence: Minimum confidence threshold (default 80)
min_loss: Minimum loss amount (default -100.0, i.e., loss >= 100)
Returns:
List of losing DecisionLog objects
"""
cursor = self.conn.execute(
"""
SELECT
decision_id, timestamp, stock_code, market, exchange_code,
action, confidence, rationale, context_snapshot, input_data,
outcome_pnl, outcome_accuracy, reviewed, review_notes
FROM decision_logs
WHERE confidence >= ?
AND outcome_pnl IS NOT NULL
AND outcome_pnl <= ?
ORDER BY outcome_pnl ASC
""",
(min_confidence, min_loss),
)
return [self._row_to_decision_log(row) for row in cursor.fetchall()]
def _row_to_decision_log(self, row: tuple[Any, ...]) -> DecisionLog:
"""Convert a database row to a DecisionLog object.
Args:
row: Database row tuple
Returns:
DecisionLog object
"""
return DecisionLog(
decision_id=row[0],
timestamp=row[1],
stock_code=row[2],
market=row[3],
exchange_code=row[4],
action=row[5],
confidence=row[6],
rationale=row[7],
context_snapshot=json.loads(row[8]),
input_data=json.loads(row[9]),
outcome_pnl=row[10],
outcome_accuracy=row[11],
reviewed=bool(row[12]),
review_notes=row[13],
)

View File

@@ -19,6 +19,7 @@ from src.broker.overseas import OverseasBroker
from src.config import Settings
from src.core.risk_manager import CircuitBreakerTripped, RiskManager
from src.db import init_db, log_trade
from src.logging.decision_logger import DecisionLogger
from src.logging_config import setup_logging
from src.markets.schedule import MarketInfo, get_next_market_open, get_open_markets
@@ -42,6 +43,7 @@ async def trading_cycle(
brain: GeminiClient,
risk: RiskManager,
db_conn: Any,
decision_logger: DecisionLogger,
market: MarketInfo,
stock_code: str,
) -> None:
@@ -101,6 +103,39 @@ async def trading_cycle(
decision.confidence,
)
# 2.5. Log decision with context snapshot
context_snapshot = {
"L1": {
"current_price": current_price,
"foreigner_net": foreigner_net,
},
"L2": {
"total_eval": total_eval,
"total_cash": total_cash,
"purchase_total": purchase_total,
"pnl_pct": pnl_pct,
},
# L3-L7 will be populated when context tree is implemented
}
input_data = {
"current_price": current_price,
"foreigner_net": foreigner_net,
"total_eval": total_eval,
"total_cash": total_cash,
"pnl_pct": pnl_pct,
}
decision_logger.log_decision(
stock_code=stock_code,
market=market.code,
exchange_code=market.exchange_code,
action=decision.action,
confidence=decision.confidence,
rationale=decision.rationale,
context_snapshot=context_snapshot,
input_data=input_data,
)
# 3. Execute if actionable
if decision.action in ("BUY", "SELL"):
# Determine order size (simplified: 1 lot)
@@ -151,6 +186,7 @@ async def run(settings: Settings) -> None:
brain = GeminiClient(settings)
risk = RiskManager(settings)
db_conn = init_db(settings.DB_PATH)
decision_logger = DecisionLogger(db_conn)
shutdown = asyncio.Event()
@@ -218,6 +254,7 @@ async def run(settings: Settings) -> None:
brain,
risk,
db_conn,
decision_logger,
market,
stock_code,
)

View File

@@ -0,0 +1,292 @@
"""Tests for decision logging and audit trail."""
from __future__ import annotations
import sqlite3
from datetime import UTC, datetime
import pytest
from src.db import init_db
from src.logging.decision_logger import DecisionLog, DecisionLogger
@pytest.fixture
def db_conn() -> sqlite3.Connection:
"""Provide an in-memory database with initialized schema."""
conn = init_db(":memory:")
return conn
@pytest.fixture
def logger(db_conn: sqlite3.Connection) -> DecisionLogger:
"""Provide a DecisionLogger instance."""
return DecisionLogger(db_conn)
def test_log_decision_creates_record(logger: DecisionLogger, db_conn: sqlite3.Connection) -> None:
"""Test that log_decision creates a database record."""
context_snapshot = {
"L1": {"quote": {"price": 100.0, "volume": 1000}},
"L2": {"orderbook": {"bid": [99.0], "ask": [101.0]}},
}
input_data = {"price": 100.0, "volume": 1000, "foreigner_net": 500}
decision_id = logger.log_decision(
stock_code="005930",
market="KR",
exchange_code="KRX",
action="BUY",
confidence=85,
rationale="Strong upward momentum",
context_snapshot=context_snapshot,
input_data=input_data,
)
# Verify decision_id is a valid UUID
assert decision_id is not None
assert len(decision_id) == 36 # UUID v4 format
# Verify record exists in database
cursor = db_conn.execute(
"SELECT decision_id, action, confidence FROM decision_logs WHERE decision_id = ?",
(decision_id,),
)
row = cursor.fetchone()
assert row is not None
assert row[0] == decision_id
assert row[1] == "BUY"
assert row[2] == 85
def test_log_decision_stores_context_snapshot(logger: DecisionLogger) -> None:
"""Test that context snapshot is stored as JSON."""
context_snapshot = {
"L1": {"real_time": "data"},
"L3": {"daily": "aggregate"},
"L7": {"legacy": "wisdom"},
}
input_data = {"price": 50000.0, "volume": 2000}
decision_id = logger.log_decision(
stock_code="035420",
market="KR",
exchange_code="KRX",
action="HOLD",
confidence=75,
rationale="Waiting for clearer signal",
context_snapshot=context_snapshot,
input_data=input_data,
)
# Retrieve and verify context snapshot
decision = logger.get_decision_by_id(decision_id)
assert decision is not None
assert decision.context_snapshot == context_snapshot
assert decision.input_data == input_data
def test_get_unreviewed_decisions(logger: DecisionLogger) -> None:
"""Test retrieving unreviewed decisions with confidence filter."""
# Log multiple decisions with varying confidence
logger.log_decision(
stock_code="005930",
market="KR",
exchange_code="KRX",
action="BUY",
confidence=90,
rationale="High confidence buy",
context_snapshot={},
input_data={},
)
logger.log_decision(
stock_code="000660",
market="KR",
exchange_code="KRX",
action="SELL",
confidence=75,
rationale="Low confidence sell",
context_snapshot={},
input_data={},
)
logger.log_decision(
stock_code="035420",
market="KR",
exchange_code="KRX",
action="HOLD",
confidence=85,
rationale="Medium confidence hold",
context_snapshot={},
input_data={},
)
# Get unreviewed decisions with default threshold (80)
unreviewed = logger.get_unreviewed_decisions()
assert len(unreviewed) == 2 # Only confidence >= 80
assert all(d.confidence >= 80 for d in unreviewed)
assert all(not d.reviewed for d in unreviewed)
# Get with lower threshold
unreviewed_all = logger.get_unreviewed_decisions(min_confidence=70)
assert len(unreviewed_all) == 3
def test_mark_reviewed(logger: DecisionLogger) -> None:
"""Test marking a decision as reviewed."""
decision_id = logger.log_decision(
stock_code="005930",
market="KR",
exchange_code="KRX",
action="BUY",
confidence=85,
rationale="Test decision",
context_snapshot={},
input_data={},
)
# Initially unreviewed
decision = logger.get_decision_by_id(decision_id)
assert decision is not None
assert not decision.reviewed
assert decision.review_notes is None
# Mark as reviewed
review_notes = "Good decision, captured bullish momentum correctly"
logger.mark_reviewed(decision_id, review_notes)
# Verify updated
decision = logger.get_decision_by_id(decision_id)
assert decision is not None
assert decision.reviewed
assert decision.review_notes == review_notes
# Should not appear in unreviewed list
unreviewed = logger.get_unreviewed_decisions()
assert all(d.decision_id != decision_id for d in unreviewed)
def test_update_outcome(logger: DecisionLogger) -> None:
"""Test updating decision outcome with P&L and accuracy."""
decision_id = logger.log_decision(
stock_code="005930",
market="KR",
exchange_code="KRX",
action="BUY",
confidence=90,
rationale="Expecting price increase",
context_snapshot={},
input_data={},
)
# Initially no outcome
decision = logger.get_decision_by_id(decision_id)
assert decision is not None
assert decision.outcome_pnl is None
assert decision.outcome_accuracy is None
# Update outcome (profitable trade)
logger.update_outcome(decision_id, pnl=5000.0, accuracy=1)
# Verify updated
decision = logger.get_decision_by_id(decision_id)
assert decision is not None
assert decision.outcome_pnl == 5000.0
assert decision.outcome_accuracy == 1
def test_get_losing_decisions(logger: DecisionLogger) -> None:
"""Test retrieving high-confidence losing decisions."""
# Profitable decision
id1 = logger.log_decision(
stock_code="005930",
market="KR",
exchange_code="KRX",
action="BUY",
confidence=85,
rationale="Correct prediction",
context_snapshot={},
input_data={},
)
logger.update_outcome(id1, pnl=3000.0, accuracy=1)
# High-confidence loss
id2 = logger.log_decision(
stock_code="000660",
market="KR",
exchange_code="KRX",
action="SELL",
confidence=90,
rationale="Wrong prediction",
context_snapshot={},
input_data={},
)
logger.update_outcome(id2, pnl=-2000.0, accuracy=0)
# Low-confidence loss (should be ignored)
id3 = logger.log_decision(
stock_code="035420",
market="KR",
exchange_code="KRX",
action="BUY",
confidence=70,
rationale="Low confidence, wrong",
context_snapshot={},
input_data={},
)
logger.update_outcome(id3, pnl=-1500.0, accuracy=0)
# Get high-confidence losing decisions
losers = logger.get_losing_decisions(min_confidence=80, min_loss=-1000.0)
assert len(losers) == 1
assert losers[0].decision_id == id2
assert losers[0].outcome_pnl == -2000.0
assert losers[0].confidence == 90
def test_get_decision_by_id_not_found(logger: DecisionLogger) -> None:
"""Test that get_decision_by_id returns None for non-existent ID."""
decision = logger.get_decision_by_id("non-existent-uuid")
assert decision is None
def test_unreviewed_limit(logger: DecisionLogger) -> None:
"""Test that get_unreviewed_decisions respects limit parameter."""
# Create 5 unreviewed decisions
for i in range(5):
logger.log_decision(
stock_code=f"00{i}",
market="KR",
exchange_code="KRX",
action="HOLD",
confidence=85,
rationale=f"Decision {i}",
context_snapshot={},
input_data={},
)
# Get only 3
unreviewed = logger.get_unreviewed_decisions(limit=3)
assert len(unreviewed) == 3
def test_decision_log_dataclass() -> None:
"""Test DecisionLog dataclass creation."""
now = datetime.now(UTC).isoformat()
log = DecisionLog(
decision_id="test-uuid",
timestamp=now,
stock_code="005930",
market="KR",
exchange_code="KRX",
action="BUY",
confidence=85,
rationale="Test",
context_snapshot={"L1": "data"},
input_data={"price": 100.0},
)
assert log.decision_id == "test-uuid"
assert log.action == "BUY"
assert log.confidence == 85
assert log.reviewed is False
assert log.outcome_pnl is None