""" 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()]