Blame for database.py

"""
database.py — SQLite 資料庫層

單一資料表 `stocks`,每檔股票一列,欄位涵蓋四面向所有指標 + 初篩用的市場資料。
提供：init_db / upsert_stock / get_symbols_needing_update / query (依 VALUE_FILTERS 動態組 SQL)。
"""

import sqlite3
import datetime as dt
from contextlib import contextmanager

import config

SCHEMA = """
CREATE TABLE IF NOT EXISTS stocks (
    symbol            TEXT PRIMARY KEY,
    company_name      TEXT,
    sector            TEXT,
    industry          TEXT,
    country           TEXT,
    exchange          TEXT,

    -- 市場 / 流動性(初篩)
    price             REAL,
    market_cap        REAL,
    avg_volume        REAL,
    dollar_volume     REAL,
    beta              REAL,

    -- 財務面
    roic              REAL,
    roic_method       TEXT,    -- 'calculated'(真實公式) 或 'roa_proxy'(退回 ROA 代理)
    debt_to_equity    REAL,
    gross_margin      REAL,
    fcf_to_revenue    REAL,

    -- 成長面
    eps_yoy           REAL,
    eps_cv            REAL,    -- EPS 波動係數,越大越像週期股
    revenue_cagr_3y   REAL,
    operating_margin  REAL,

    -- 估值面
    pe                REAL,
    forward_pe        REAL,
    peg               REAL,
    ev_ebitda         REAL,

    -- 市場面
    volatility_annual REAL,
    momentum_12_1     REAL,

    passes_universe   INTEGER DEFAULT 0,   -- 是否通過初篩(含 dollar_volume)
    data_source       TEXT,
    updated_at        TEXT,

    -- 學術公開公式評分(Altman Z-Score / Piotroski F-Score / Mohanram G-Score)
    altman_z_score       REAL,    -- Altman(1968)破產風險評分;金融股不適用
    altman_zone          TEXT,    -- 'safe' / 'grey' / 'distress'
    piotroski_f_score    INTEGER, -- Piotroski(2000)財務體質評分,實得分數
    piotroski_max        INTEGER, -- 實際可計算的子項數(滿分基準,可能 <9)
    mohanram_roa            REAL, -- 以下為 Mohanram G-Score 原始比率(未跟同業比較前)
    mohanram_cfo_to_assets  REAL,
    mohanram_cfo_gt_ni       INTEGER,  -- 0/1,布林值
    mohanram_rd_to_assets   REAL,
    mohanram_capex_to_assets REAL,
    mohanram_roa_cv         REAL,      -- 多年 ROA 波動係數(公司內部,不需同業比較)
    mohanram_sales_growth_cv REAL,     -- 多年營收成長波動係數(公司內部,不需同業比較)
    mohanram_g_score        INTEGER,   -- 依同業中位數比較後的最終分數(0-7),由 pipeline 第二輪回填
    mohanram_g_max          INTEGER    -- 實際可計算的子項數(滿分基準,可能 <7)
);
CREATE INDEX IF NOT EXISTS idx_passes ON stocks(passes_universe);
CREATE INDEX IF NOT EXISTS idx_updated ON stocks(updated_at);
"""

_COLUMNS = [
    "symbol", "company_name", "sector", "industry", "country", "exchange",
    "price", "market_cap", "avg_volume", "dollar_volume", "beta",
    "roic", "roic_method", "debt_to_equity", "gross_margin", "fcf_to_revenue",
    "eps_yoy", "eps_cv", "revenue_cagr_3y", "operating_margin",
    "pe", "forward_pe", "peg", "ev_ebitda",
    "volatility_annual", "momentum_12_1",
    "passes_universe", "data_source", "updated_at",
    "altman_z_score", "altman_zone", "piotroski_f_score", "piotroski_max",
    "mohanram_roa", "mohanram_cfo_to_assets", "mohanram_cfo_gt_ni",
    "mohanram_rd_to_assets", "mohanram_capex_to_assets",
    "mohanram_roa_cv", "mohanram_sales_growth_cv",
    "mohanram_g_score", "mohanram_g_max",
]

# 為了向後相容舊資料庫檔(手動測試留下的),補齊缺少的欄位
_MIGRATIONS = {
    "roic_method": "ALTER TABLE stocks ADD COLUMN roic_method TEXT",
    "eps_cv": "ALTER TABLE stocks ADD COLUMN eps_cv REAL",
    "altman_z_score": "ALTER TABLE stocks ADD COLUMN altman_z_score REAL",
    "altman_zone": "ALTER TABLE stocks ADD COLUMN altman_zone TEXT",
    "piotroski_f_score": "ALTER TABLE stocks ADD COLUMN piotroski_f_score INTEGER",
    "piotroski_max": "ALTER TABLE stocks ADD COLUMN piotroski_max INTEGER",
    "mohanram_roa": "ALTER TABLE stocks ADD COLUMN mohanram_roa REAL",
    "mohanram_cfo_to_assets": "ALTER TABLE stocks ADD COLUMN mohanram_cfo_to_assets REAL",
    "mohanram_cfo_gt_ni": "ALTER TABLE stocks ADD COLUMN mohanram_cfo_gt_ni INTEGER",
    "mohanram_rd_to_assets": "ALTER TABLE stocks ADD COLUMN mohanram_rd_to_assets REAL",
    "mohanram_capex_to_assets": "ALTER TABLE stocks ADD COLUMN mohanram_capex_to_assets REAL",
    "mohanram_roa_cv": "ALTER TABLE stocks ADD COLUMN mohanram_roa_cv REAL",
    "mohanram_sales_growth_cv": "ALTER TABLE stocks ADD COLUMN mohanram_sales_growth_cv REAL",
    "mohanram_g_score": "ALTER TABLE stocks ADD COLUMN mohanram_g_score INTEGER",
    "mohanram_g_max": "ALTER TABLE stocks ADD COLUMN mohanram_g_max INTEGER",
}


@contextmanager
def connect(db_path=None):
    conn = sqlite3.connect(db_path or config.DB_PATH)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
        conn.commit()
    finally:
        conn.close()


def _ensure_columns(conn):
    existing = {row["name"] for row in conn.execute("PRAGMA table_info(stocks)")}
    for col, sql in _MIGRATIONS.items():
        if col not in existing:
            conn.execute(sql)


def init_db(db_path=None):
    with connect(db_path) as conn:
        conn.executescript(SCHEMA)
        _ensure_columns(conn)


def upsert_stock(record: dict, db_path=None):
    """record 為 dict,缺的欄位以 None 填。symbol 必填。"""
    record = {**record}
    record["updated_at"] = dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds")
    row = [record.get(c) for c in _COLUMNS]
    placeholders = ",".join("?" * len(_COLUMNS))
    updates = ",".join(f"{c}=excluded.{c}" for c in _COLUMNS if c != "symbol")
    sql = (
        f"INSERT INTO stocks ({','.join(_COLUMNS)}) VALUES ({placeholders}) "
        f"ON CONFLICT(symbol) DO UPDATE SET {updates}"
    )
    with connect(db_path) as conn:
        conn.execute(sql, row)


def get_symbols_needing_update(db_path=None, stale_days=None):
    """回傳尚未抓 / 資料過期的 symbol(給分批補資料用)。"""
    stale_days = stale_days if stale_days is not None else config.REFRESH_STALE_AFTER_DAYS
    cutoff = (dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=stale_days)).isoformat()
    with connect(db_path) as conn:
        rows = conn.execute(
            "SELECT symbol FROM stocks WHERE updated_at IS NULL OR updated_at < ? "
            "ORDER BY updated_at IS NOT NULL, updated_at ASC",
            (cutoff,),
        ).fetchall()
    return [r["symbol"] for r in rows]


def seed_universe(symbols, db_path=None):
    """把候選 symbol 先寫進去(其餘欄位待 enrich)。
    passes_universe 預設 0,等 enrich 拿到實際資料後才決定是否為 1,
    避免種子階段就被誤判為「通過初篩」。"""
    with connect(db_path) as conn:
        for s in symbols:
            conn.execute(
                "INSERT OR IGNORE INTO stocks (symbol, passes_universe) VALUES (?, 0)", (s,)
            )


def list_sectors(db_path=None):
    """回傳資料庫裡目前有的所有 sector(供 UI 下拉選單動態產生選項)。"""
    with connect(db_path) as conn:
        rows = conn.execute(
            "SELECT DISTINCT sector FROM stocks WHERE sector IS NOT NULL ORDER BY sector"
        ).fetchall()
    return [r["sector"] for r in rows]


def get_mohanram_raw_data(db_path=None):
    """
    取出所有通過初篩、且有 sector 的股票的 Mohanram 原始比率,
    供 pipeline 依 sector 分組算中位數、回填最終 0-7 分用。
    """
    with connect(db_path) as conn:
        rows = conn.execute(
            "SELECT symbol, sector, mohanram_roa, mohanram_cfo_to_assets, "
            "mohanram_cfo_gt_ni, mohanram_roa_cv, mohanram_sales_growth_cv, "
            "mohanram_rd_to_assets, mohanram_capex_to_assets "
            "FROM stocks WHERE passes_universe=1 AND sector IS NOT NULL"
        ).fetchall()
    return [dict(r) for r in rows]


def update_mohanram_score(symbol, score, max_n, db_path=None):
    """只更新 mohanram_g_score / mohanram_g_max,不動其他欄位。"""
    with connect(db_path) as conn:
        conn.execute(
            "UPDATE stocks SET mohanram_g_score=?, mohanram_g_max=? WHERE symbol=?",
            (score, max_n, symbol),
        )


def query(filters=None, ops=None, require_universe=True, order_by="roic DESC",
          market_cap_min=None, market_cap_max=None, sectors=None, db_path=None):
    """
    依 VALUE_FILTERS 動態組 WHERE。None 的條件自動略過。
    只比較「有資料」的列(該欄位 NOT NULL),避免缺值被誤判為通過。

    market_cap_min / market_cap_max:市值區間(單一欄位的上下限,VALUE_FILTERS
    的機制只支援單一比較方向,無法同時做 >= 又 <=,所以獨立成參數)。
    sectors:list[str],限定產業別(多選);None 或空 list 代表不限制。
    這兩個參數用來重現「市值介於 A-B + 限定產業」這類 scope-narrowing 策略
    (例如中型科技股策略),跟 VALUE_FILTERS 的四面向品質門檻是兩個獨立的層次。
    """
    filters = config.VALUE_FILTERS if filters is None else filters
    ops = config.VALUE_FILTER_OPS if ops is None else ops

    where, params = [], []
    if require_universe:
        where.append("passes_universe = 1")

    if market_cap_min is not None:
        where.append("market_cap IS NOT NULL AND market_cap >= ?")
        params.append(market_cap_min)
    if market_cap_max is not None:
        where.append("market_cap IS NOT NULL AND market_cap <= ?")
        params.append(market_cap_max)

    if sectors:
        placeholders = ",".join("?" * len(sectors))
        where.append(f"sector IN ({placeholders})")
        params.extend(sectors)

    for col, thr in filters.items():
        if thr is None:
            continue
        op = ops.get(col, ">=")
        where.append(f"({col} IS NOT NULL AND {col} {op} ?)")
        params.append(thr)

    sql = "SELECT * FROM stocks"
    if where:
        sql += " WHERE " + " AND ".join(where)
    if order_by:
        sql += f" ORDER BY {order_by}"
    with connect(db_path) as conn:
        return [dict(r) for r in conn.execute(sql, params).fetchall()]