Spaces:
Sleeping
Sleeping
| """ | |
| 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", | |
| } | |
| 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()] | |