crypto_news/database/models.py

"""Database models and schema management for SQLite.

Provides dataclass-based models and automatic table creation.
"""

from __future__ import annotations

import json
import sqlite3
from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Optional

from loguru import logger

from config import settings


# ------------------------------------------------------------------
# SQL Schema
# ------------------------------------------------------------------

SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS positions (
    id TEXT PRIMARY KEY,
    symbol TEXT NOT NULL,
    direction TEXT NOT NULL,
    entry_price REAL NOT NULL,
    amount REAL NOT NULL,
    stop_loss REAL NOT NULL,
    take_profit REAL NOT NULL,
    trailing_stop REAL,
    realized_pnl REAL DEFAULT 0,
    status TEXT DEFAULT 'OPEN',
    opened_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    closed_at TIMESTAMP,
    close_reason TEXT,
    confluence_score INTEGER,
    entry_reasons TEXT
);

CREATE TABLE IF NOT EXISTS trade_records (
    id TEXT PRIMARY KEY,
    position_id TEXT REFERENCES positions(id),
    symbol TEXT NOT NULL,
    side TEXT NOT NULL,
    order_type TEXT NOT NULL,
    price REAL NOT NULL,
    amount REAL NOT NULL,
    fee REAL DEFAULT 0,
    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE IF NOT EXISTS daily_performance (
    date TEXT PRIMARY KEY,
    total_trades INTEGER DEFAULT 0,
    winning_trades INTEGER DEFAULT 0,
    losing_trades INTEGER DEFAULT 0,
    total_pnl REAL DEFAULT 0,
    max_drawdown REAL DEFAULT 0
);

CREATE TABLE IF NOT EXISTS bot_state (
    key TEXT PRIMARY KEY,
    value TEXT,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
"""


def init_db(db_path: str | None = None) -> sqlite3.Connection:
    """Create the database and tables if they do not exist."""
    path = db_path or settings.DB_PATH
    conn = sqlite3.connect(path, check_same_thread=False)
    conn.executescript(SCHEMA_SQL)
    conn.commit()
    logger.info("Database initialised at {}", path)
    return conn


# ------------------------------------------------------------------
# Data Models (mirrors of DB rows)
# ------------------------------------------------------------------

@dataclass
class PositionRecord:
    """DB-level position record."""

    id: str
    symbol: str
    direction: str
    entry_price: float
    amount: float
    stop_loss: float
    take_profit: float
    trailing_stop: Optional[float] = None
    realized_pnl: float = 0.0
    status: str = "OPEN"
    opened_at: str = ""
    closed_at: Optional[str] = None
    close_reason: Optional[str] = None
    confluence_score: int = 0
    entry_reasons: str = "[]"


@dataclass
class TradeRecord:
    """DB-level trade record."""

    id: str
    position_id: str
    symbol: str
    side: str
    order_type: str
    price: float
    amount: float
    fee: float = 0.0
    timestamp: str = ""


@dataclass
class DailyPerformance:
    """DB-level daily performance summary."""

    date: str
    total_trades: int = 0
    winning_trades: int = 0
    losing_trades: int = 0
    total_pnl: float = 0.0
    max_drawdown: float = 0.0

    @property
    def win_rate(self) -> float:
        if self.total_trades == 0:
            return 0.0
        return self.winning_trades / self.total_trades

    @property
    def losing_rate(self) -> float:
        if self.total_trades == 0:
            return 0.0
        return self.losing_trades / self.total_trades