Database Schema 文檔

概述

Cortex Investment Agent 使用 PostgreSQL 作為主要資料庫，儲存股票資料、技術指標和 Agent 執行記錄。

技術選擇: - PostgreSQL 14+ - SQLAlchemy 2.0 (ORM) - Alembic (資料庫遷移)

資料庫設計

ER Diagram

┌─────────────┐
│   stocks    │
└──────┬──────┘
       │ 1
       │
       │ N
┌──────▼──────┐         ┌──────────────────────┐
│ stock_ohlc  │         │ technical_indicators │
└─────────────┘         └──────────────────────┘

┌──────────────────┐
│ agent_executions │
└──────────────────┘

表結構定義

1. `taiwan.stocks`／`us.stocks`（股票基本資料）

用途: 依國別實體隔離股票基本資訊。DB 內部只保存裸 stock_code；台股的 listing_type 為 tw／two，指數與美股為 NULL。對外 API 仍使用 2330.TW 等既有 ticker 格式。

欄位:

CREATE SCHEMA taiwan;
CREATE SCHEMA us;

CREATE TABLE taiwan.stocks (
    stock_code VARCHAR(20) PRIMARY KEY,
    listing_type VARCHAR(8),
    name VARCHAR(200),
    industry VARCHAR(100),
    sector VARCHAR(100),
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW(),
    CHECK (listing_type IN ('tw', 'two'))
);

CREATE TABLE us.stocks (
    LIKE taiwan.stocks INCLUDING DEFAULTS
);
-- 實際 migration 另建立 country-local constraints；US listing_type 必須為 NULL。

Python Model:

# backend/database/models.py

from sqlalchemy import Column, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

class TaiwanStock(Base):
    __tablename__ = "stocks"
    __table_args__ = {"schema": "taiwan"}

    stock_code = Column(String(20), primary_key=True)
    listing_type = Column(String(8))
    name = Column(String(200))
    industry = Column(String(100))
    sector = Column(String(100))
    created_at = Column(DateTime, default=datetime.now)
    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)

範例資料:

INSERT INTO taiwan.stocks (stock_code, listing_type, name, industry, sector)
VALUES ('2330', 'tw', '台積電', '半導體', '科技');

INSERT INTO us.stocks (stock_code, listing_type, name, industry, sector)
VALUES ('AAPL', NULL, 'Apple Inc.', '消費性電子', '科技');

2. `taiwan.stock_ohlc`／`us.stock_ohlc`（股票 OHLC 時序資料）

用途: 儲存每日 OHLC (Open, High, Low, Close) 和成交量

欄位:

CREATE TABLE taiwan.stock_ohlc (
    stock_code VARCHAR(20) NOT NULL,
    listing_type VARCHAR(8), -- 日期有效的歷史上市別，不參與 identity/FK
    timestamp TIMESTAMP NOT NULL,
    open NUMERIC(12, 4) NOT NULL,
    high NUMERIC(12, 4) NOT NULL,
    low NUMERIC(12, 4) NOT NULL,
    close NUMERIC(12, 4) NOT NULL,
    volume BIGINT NOT NULL,
    PRIMARY KEY (stock_code, timestamp),
    FOREIGN KEY (stock_code) REFERENCES taiwan.stocks(stock_code) ON DELETE CASCADE
);

-- 索引（加速查詢）
CREATE INDEX idx_ohlc_stock_code_time
    ON taiwan.stock_ohlc(stock_code, timestamp DESC);
CREATE INDEX idx_ohlc_timestamp ON taiwan.stock_ohlc(timestamp DESC);

Python Model:

from sqlalchemy import Column, String, DateTime, Numeric, BigInteger, ForeignKey

class TaiwanStockOHLC(Base):
    __tablename__ = "stock_ohlc"
    __table_args__ = {"schema": "taiwan"}

    stock_code = Column(
        String(20), ForeignKey("taiwan.stocks.stock_code"), primary_key=True
    )
    listing_type = Column(String(8))
    timestamp = Column(DateTime, primary_key=True)
    open = Column(Numeric(12, 4), nullable=False)
    high = Column(Numeric(12, 4), nullable=False)
    low = Column(Numeric(12, 4), nullable=False)
    close = Column(Numeric(12, 4), nullable=False)
    volume = Column(BigInteger, nullable=False)

查詢範例:

-- 查詢台積電最近 30 天資料
SELECT * FROM taiwan.stock_ohlc
WHERE stock_code = '2330'
  AND timestamp >= CURRENT_DATE - INTERVAL '30 days'
ORDER BY timestamp DESC;

-- 查詢特定日期所有股票收盤價
SELECT ticker, close 
FROM stock_ohlc
WHERE timestamp = '2025-01-31';

3. technical_indicators (技術指標)

用途: 儲存預先計算的技術指標

欄位:

CREATE TABLE technical_indicators (
    ticker VARCHAR(20) NOT NULL,
    timestamp TIMESTAMP NOT NULL,

    -- 移動平均線（每個 ma_<N> 為真實 N 日 SMA，無 alias）
    ma_5 NUMERIC(12, 4),
    ma_10 NUMERIC(12, 4),
    ma_20 NUMERIC(12, 4),
    ma_50 NUMERIC(12, 4),
    ma_60 NUMERIC(12, 4),    -- TW 季線
    ma_100 NUMERIC(12, 4),
    ma_120 NUMERIC(12, 4),   -- TW 半年線
    ma_200 NUMERIC(12, 4),
    ma_240 NUMERIC(12, 4),   -- TW 年線

    -- 指數移動平均線
    ema_5 NUMERIC(12, 4),
    ema_10 NUMERIC(12, 4),
    ema_20 NUMERIC(12, 4),
    ema_50 NUMERIC(12, 4),
    ema_60 NUMERIC(12, 4),    -- TW 季線
    ema_100 NUMERIC(12, 4),
    ema_120 NUMERIC(12, 4),   -- TW 半年線
    ema_200 NUMERIC(12, 4),
    ema_240 NUMERIC(12, 4),   -- TW 年線

    -- RSI
    rsi NUMERIC(6, 2),

    -- MACD
    macd NUMERIC(12, 4),
    macd_signal NUMERIC(12, 4),
    macd_hist NUMERIC(12, 4),

    -- KD
    k NUMERIC(6, 2),
    d NUMERIC(6, 2),

    -- 布林通道
    bb_upper NUMERIC(12, 4),
    bb_middle NUMERIC(12, 4),
    bb_lower NUMERIC(12, 4),

    -- ATR (真實波動幅度)
    atr NUMERIC(12, 4),

    -- ADX (趨勢強度)
    adx NUMERIC(6, 2),

    -- OBV (能量潮)
    obv BIGINT,

    -- William %R
    willr NUMERIC(6, 2),

    -- 成交量均線
    volume_ma_5 BIGINT,
    volume_ma_20 BIGINT,

    -- VWAP
    vwap NUMERIC(12, 4),

    -- Ichimoku
    ichimoku_tenkan NUMERIC(12, 4),
    ichimoku_kijun NUMERIC(12, 4),

    -- CCI
    cci NUMERIC(8, 2),

    -- MFI
    mfi NUMERIC(6, 2),

    PRIMARY KEY (stock_code, timestamp),
    FOREIGN KEY (stock_code) REFERENCES taiwan.stocks(stock_code) ON DELETE CASCADE
);

-- 索引
CREATE INDEX idx_indicators_stock_code_time
    ON technical_indicators(stock_code, timestamp DESC);

Python Model:

class TechnicalIndicator(Base):
    __tablename__ = "technical_indicators"

    stock_code = Column(
        String(20), ForeignKey("taiwan.stocks.stock_code"), primary_key=True
    )
    timestamp = Column(DateTime, primary_key=True)

    # 移動平均線（每個 ma_<N> 為真實 N 日 SMA，無 alias）
    ma_5 = Column(Numeric(12, 4))
    ma_10 = Column(Numeric(12, 4))
    ma_20 = Column(Numeric(12, 4))
    ma_50 = Column(Numeric(12, 4))
    ma_60 = Column(Numeric(12, 4))   # TW 季線
    ma_100 = Column(Numeric(12, 4))
    ma_120 = Column(Numeric(12, 4))  # TW 半年線
    ma_200 = Column(Numeric(12, 4))
    ma_240 = Column(Numeric(12, 4))  # TW 年線

    # EMA
    ema_5 = Column(Numeric(12, 4))
    ema_10 = Column(Numeric(12, 4))
    ema_20 = Column(Numeric(12, 4))
    ema_50 = Column(Numeric(12, 4))
    ema_60 = Column(Numeric(12, 4))   # TW 季線
    ema_100 = Column(Numeric(12, 4))
    ema_120 = Column(Numeric(12, 4))  # TW 半年線
    ema_200 = Column(Numeric(12, 4))
    ema_240 = Column(Numeric(12, 4))  # TW 年線

    # RSI
    rsi = Column(Numeric(6, 2))

    # MACD
    macd = Column(Numeric(12, 4))
    macd_signal = Column(Numeric(12, 4))
    macd_hist = Column(Numeric(12, 4))

    # KD
    k = Column(Numeric(6, 2))
    d = Column(Numeric(6, 2))

    # Bollinger Bands
    bb_upper = Column(Numeric(12, 4))
    bb_middle = Column(Numeric(12, 4))
    bb_lower = Column(Numeric(12, 4))

    # 其他指標...
    atr = Column(Numeric(12, 4))
    adx = Column(Numeric(6, 2))
    obv = Column(BigInteger)
    willr = Column(Numeric(6, 2))
    volume_ma_5 = Column(BigInteger)
    volume_ma_20 = Column(BigInteger)
    vwap = Column(Numeric(12, 4))
    ichimoku_tenkan = Column(Numeric(12, 4))
    ichimoku_kijun = Column(Numeric(12, 4))
    cci = Column(Numeric(8, 2))
    mfi = Column(Numeric(6, 2))

查詢範例:

-- 查詢台積電最新指標
SELECT * FROM technical_indicators
WHERE ticker = '2330.TW'
ORDER BY timestamp DESC
LIMIT 1;

-- 查詢 RSI 超買的股票（>70）
SELECT ticker, timestamp, rsi, close
FROM technical_indicators ti
JOIN stock_ohlc so USING (ticker, timestamp)
WHERE timestamp = (SELECT MAX(timestamp) FROM technical_indicators)
  AND rsi > 70;

4. agent_executions (Agent 執行記錄)

用途: 記錄每次 Agent 執行的詳細資訊，用於監控和優化

欄位:

CREATE TABLE agent_executions (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    agent_name VARCHAR(50) NOT NULL,
    task_type VARCHAR(50) NOT NULL,

    -- 任務輸入
    task_input JSONB NOT NULL,

    -- Meta-Agent 生成的策略
    strategy JSONB,

    -- 執行結果
    result JSONB,

    -- VRR 狀態
    verification_passed BOOLEAN,
    iterations INTEGER DEFAULT 0,

    -- 效能指標
    execution_time_ms INTEGER,

    -- 錯誤資訊
    error TEXT,

    -- 時間戳
    created_at TIMESTAMP DEFAULT NOW()
);

-- 索引
CREATE INDEX idx_executions_agent ON agent_executions(agent_name);
CREATE INDEX idx_executions_created_at ON agent_executions(created_at DESC);
CREATE INDEX idx_executions_verification ON agent_executions(verification_passed);

Python Model:

from sqlalchemy.dialects.postgresql import UUID, JSONB
import uuid

class AgentExecution(Base):
    __tablename__ = "agent_executions"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    agent_name = Column(String(50), nullable=False, index=True)
    task_type = Column(String(50), nullable=False)

    task_input = Column(JSONB, nullable=False)
    strategy = Column(JSONB)
    result = Column(JSONB)

    verification_passed = Column(Boolean, index=True)
    iterations = Column(Integer, default=0)
    execution_time_ms = Column(Integer)
    error = Column(String)

    created_at = Column(DateTime, default=datetime.now, index=True)

範例資料:

{
  "id": "550e8400-e29b-41d4-a716-446655440000",
  "agent_name": "stock_analyzer",
  "task_type": "comprehensive_analysis",
  "task_input": {
    "ticker": "2330.TW",
    "analysis_type": "comprehensive",
    "timeframe": "3_months"
  },
  "strategy": {
    "verify_criteria": [
      "RSI 數值在 0-100 範圍內",
      "MACD 交叉點識別正確"
    ],
    "reflect_focus": [
      "技術指標互相印證性",
      "風險提示完整性"
    ],
    "refine_priority": [
      "加強數據支持",
      "明確化建議"
    ]
  },
  "result": {
    "summary": "台積電當前處於中性偏多...",
    "trend": "bullish",
    "confidence": 0.75
  },
  "verification_passed": true,
  "iterations": 1,
  "execution_time_ms": 3250,
  "created_at": "2025-02-01T10:30:45"
}

分析查詢:

-- 計算各 Agent 成功率
SELECT 
    agent_name,
    COUNT(*) as total,
    SUM(CASE WHEN verification_passed THEN 1 ELSE 0 END) as passed,
    ROUND(AVG(CASE WHEN verification_passed THEN 1.0 ELSE 0.0 END) * 100, 2) as success_rate
FROM agent_executions
WHERE created_at >= CURRENT_DATE - INTERVAL '7 days'
GROUP BY agent_name;

-- 常見失敗任務
SELECT 
    task_type,
    task_input->>'ticker' as ticker,
    COUNT(*) as failures
FROM agent_executions
WHERE verification_passed = FALSE
  AND created_at >= CURRENT_DATE - INTERVAL '7 days'
GROUP BY task_type, task_input->>'ticker'
ORDER BY failures DESC
LIMIT 10;

-- 效能分析
SELECT 
    agent_name,
    AVG(execution_time_ms) as avg_time,
    MAX(execution_time_ms) as max_time,
    PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY execution_time_ms) as p95_time
FROM agent_executions
WHERE created_at >= CURRENT_DATE - INTERVAL '7 days'
GROUP BY agent_name;

Token session monitoring (`refresh_tokens`)

refresh_tokens 只保存 refresh token 的 SHA-256 hash，不保存 refresh token 或 access bearer token 明文。Migration 116_refresh_token_access_expiry.sql 新增：

access_token_expires_at TIMESTAMPTZ NOT NULL

此欄位保存與該 refresh-token session 同時簽發之 access JWT 的精確到期時間，供管理端以 token 有效性計算目前上線人數。登入與 token rotation 必須在同一交易中寫入新的 access_token_expires_at；JWT 的 exp 與資料庫欄位共用同一個整秒 UTC 時間，避免小數秒造成狀態不一致。

歷史資料依原有 60 分鐘 access-token policy 由 created_at + INTERVAL '60 minutes' 回填。若 legacy row 缺少 created_at，migration 會 fail closed，不會自行虛構到期時間。

CREATE INDEX idx_refresh_tokens_access_expiry
    ON refresh_tokens (access_token_expires_at)
    WHERE revoked_at IS NULL;

CREATE INDEX idx_refresh_tokens_created_at
    ON refresh_tokens (created_at);

第一個索引支援目前有效 session 查詢；第二個索引支援今日、本週、本月與自訂日期區間的 token 簽發統計。目前上線 session 還必須同時符合 revoked_at IS NULL、refresh token 尚未到期、session 已建立，以及所屬會員狀態為 active。

Access-log 真實起訖時間 (`api_access_log`)

Migration 117_access_log_request_response_timestamps.sql 新增兩個 nullable 欄位：

request_started_at      TIMESTAMPTZ NULL,
response_completed_at  TIMESTAMPTZ NULL

request_started_at：middleware 開始觀察 request 的 UTC wall-clock 時間。
response_completed_at：final response body 成功送出後的 UTC wall-clock 時間。
既有 timestamp：背景 writer 建立／持久化 row 的時間，不代表 request start 或 response completion。

歷史資料不做推算回填，兩個新欄位維持 NULL；未送出 final response body 的例外紀錄也可能沒有 response_completed_at。不得以 timestamp ± duration_ms 寫回或宣稱為真實起訖時間。

Theme recap 快取版本與 access-log 細部耗時

Migration 146_theme_recap_response_cache_versions.sql 在既有 public.response_cache_data_version registry 新增五個 theme_recap.* row，分別對應 coverage_theme_membership、strategy_match、 daily_chip_stock_event、daily_technical_stock_event 與 margin_balance； stocks、stock_ohlc 則沿用 migrations 114/134 已有的 global row 與 trigger。每張來源表的 statement-level DML／TRUNCATE trigger 只旋轉自己的版本，因此不同來源的長交易不會爭用同一 registry tuple lock；rollback 會連同版本更新一起回滾。API 以固定七欄 registry query 取得完整向量，七個 counter 各自寫入 cache key，不得以 MAX／GREATEST 合併。若缺少任一 row，API 會 fail loud 並要求先套用 migration 146，不會以 0 猜測版本。registry 不承載題材業務資料。

Migration 147_api_access_log_timing_breakdown.sql 對 api_access_log 新增：

db_duration_ms   NUMERIC(12, 2) NULL,
db_query_count   INTEGER NULL,
timing_breakdown JSONB NULL

三欄均無 default、不回填歷史 row。新 HTTP row 的 db_duration_ms 是 request 在 API engine 上的 SQL cursor execution 累計，db_query_count 是相同範圍的 query 次數；背景 access-log INSERT 不在計算內。timing_breakdown 使用 schema_version = 1，包含 allowlisted phases_ms 與 named cache_outcomes。DB 累計時間可能和 phase 重疊或因並行 SQL 大於 wall-clock，不可與 phase 直接相加。#2056 起，新 row 一律含附加 key db_pool； db_pool_spawned_tasks、sync_cpu、post_response 只在有內容時才寫入（連線取得／持有與同步 CPU 量測；schema_version 不變、無 migration），定義見 docs/runbooks/api-runtime-metrics.md。

Theme store（題材正典）：`public.theme` / `taiwan.theme_member`

Migration 151_theme_store.sql 新增題材正典的兩張表，ORM 對應 ThemeStoreTheme（public.theme）與 TaiwanThemeStoreMember （taiwan.theme_member）。API 規格見 docs/api-reference.md「Theme Store （題材正典 CRUD）」一節。

CREATE TABLE IF NOT EXISTS public.theme (
    theme_name   VARCHAR(200) NOT NULL,
    description  TEXT NULL,
    source       VARCHAR(16) NOT NULL,
    is_edited    BOOLEAN NOT NULL DEFAULT FALSE,
    is_blocked   BOOLEAN NOT NULL DEFAULT FALSE,
    created_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
    CONSTRAINT pk_theme PRIMARY KEY (theme_name),
    CONSTRAINT chk_theme_source CHECK (source IN ('manual', 'coverage'))
);

CREATE TABLE IF NOT EXISTS taiwan.theme_member (
    theme_name   VARCHAR(200) NOT NULL,
    stock_code   VARCHAR(20)  NOT NULL,
    company_name TEXT NULL,
    listing_type VARCHAR(8)   NOT NULL,
    role         VARCHAR(16)  NULL,
    source       VARCHAR(16)  NOT NULL,
    created_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
    CONSTRAINT pk_theme_member PRIMARY KEY (theme_name, stock_code),
    CONSTRAINT fk_theme_member_theme
        FOREIGN KEY (theme_name) REFERENCES public.theme (theme_name) ON DELETE CASCADE,
    CONSTRAINT chk_theme_member_stock_code_bare
        CHECK (upper(stock_code) NOT LIKE '%.TW' AND upper(stock_code) NOT LIKE '%.TWO'),
    CONSTRAINT chk_theme_member_listing_type
        CHECK (listing_type IN ('tw', 'two')),
    CONSTRAINT chk_theme_member_role
        CHECK (role IS NULL OR role IN ('upstream', 'midstream', 'downstream', 'related')),
    CONSTRAINT chk_theme_member_source
        CHECK (source IN ('manual', 'coverage'))
);

CREATE INDEX IF NOT EXISTS idx_theme_member_stock_code ON taiwan.theme_member (stock_code);

為什麼沒有 knowledge_as_of：新表是 current-state 的正典題材表，不是 coverage 那種「每次 sync 依快照日期整批刪光重灌」的語料鏡射表。本次遷移涉及的讀取點（Phase 1 的 GET /api/themes、GET /api/themes/{id}/stocks，以及 Phase 2 候選 theme_recap_service、stock_lists、chip_recap_service、 capital_flow_diff_service、coverage_graph_service 等）查詢舊表 taiwan.coverage_theme_membership 時全部只取 MAX(knowledge_as_of) 之後的最新快照，從未依歷史快照查詢；因此新表不需要這個維度，讀取邏輯也因此變簡單。

為什麼 listing_type 要放在成員列：GET /api/themes/{id}/stocks 用 COALESCE(m.listing_type, stock.listing_type) 決定每個成員的掛牌類別，再交給 format_external_ticker 組出對外 ticker；format_external_ticker 在 listing_type 為 None 時回傳沒有 .TW/.TWO 後綴的裸代號，不是報錯。也就是說成員列自己帶的 listing_type 才是主要來源，taiwan.stocks 只是 fallback——股票不在 taiwan.stocks（例如已被清理或尚未回填）時，唯一能還原正確 ticker 後綴的依據就是成員列自己的 listing_type。這欄若漏掉，wrapper 匯入的成員只要剛好不在 taiwan.stocks，對外 ticker 就會靜默少掉 .TW/.TWO 後綴，不會報錯、也不會被既有測試攔下。

為什麼 listing_type 是 NOT NULL：CountryStockIdentity.__post_init__ 對台股代號（非 ^ 開頭的指數代號）在 listing_type 為 None 時直接 raise（investment-agent/backend/services/stock_identity.py:98-101）；一旦成員列存進 NULL，GET /api/themes/{id}/stocks 建構該成員的 CountryStockIdentity 時就會整支 500，拖垮整個題材而不只是那一筆成員。 taiwan.coverage_theme_membership 的既有 CHECK 約束（ck_coverage_theme_membership_stock_code_listing__723894ae）本來就對非 NULL、非 ^ 開頭的 stock_code 強制 listing_type 必須是 tw/two；本表的 NOT NULL 只是把同一條規則落實到欄位層級。因此兩條寫入路徑都必須在 INSERT 前解析出合法的 listing_type，解析不出來時分別是拒絕寫入（手動路徑，422）或跳過該成員（POST /import-from-coverage，見 docs/api-reference.md「Theme Store」一節），沒有第三種「先存 NULL 再說」的路徑。

is_edited / is_blocked 語意：

is_edited：該題材是否被人工（透過 /api/admin/theme-store 的寫入端點）編輯過。coverage 來源的題材一旦 is_edited=true，就永久跳出 POST /import-from-coverage 的覆蓋範圍，直到走還原路徑。全新建立的手動題材固定 is_edited=false。
is_blocked：該題材是否已被使用者刪除（墓碑列）。is_blocked=true 的題材對所有讀取端點視同不存在（詳見 docs/api-reference.md「is_blocked 可見性規則」）。
刪除＝完整狀態重置：DELETE /api/admin/theme-store/themes/{theme_name} 會清空該題材的所有成員、把 is_blocked 設為 true，同時把 is_edited 重設為 false，只保留 theme_name 與 source。這是必要設計，不是附帶行為：「刪除 → DELETE /blocked/{name} 允許重新接入 → 重跑 POST /import-from-coverage」這條還原路徑，如果 is_edited 沒有一併重置， unblock 後 wrapper 仍會因為 is_edited=true 而繼續跳過該題材（見 docs/api-reference.md「POST /import-from-coverage」判斷順序第 3 條），結果會停在一個「可見、成員清空、wrapper 永遠不會再碰」的殭屍狀態。

taiwan.coverage_theme_membership 完全未受影響：舊表 schema、資料、寫入路徑本次一律不動，仍由既有 CoverageETL 依 knowledge_as_of 快照餵資料。 Phase 1 只把 GET /api/themes 與 GET /api/themes/{id}/stocks 兩支端點切到新表讀取；讀舊表的其餘服務（theme_recap_service、coverage_graph_service、 stock_lists、chip_recap_service、capital_flow_diff_service 等）留給 Phase 2 遷移。

資料庫初始化

init_db.py

檔案: backend/database/init_db.py

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from database.models import Base
import logging

logger = logging.getLogger(__name__)

async def init_database(database_url: str):
    """初始化資料庫"""

    # 建立引擎
    engine = create_async_engine(
        database_url,
        echo=False,
        pool_pre_ping=True,
        pool_size=10,
        max_overflow=20
    )

    # 建立所有表
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    logger.info("資料庫初始化完成")

    return engine

async def get_session(engine):
    """獲取資料庫 session"""
    async_session = sessionmaker(
        engine,
        class_=AsyncSession,
        expire_on_commit=False
    )

    async with async_session() as session:
        yield session

CRUD 操作

crud.py

檔案: backend/database/crud.py

from sqlalchemy import select, and_
from sqlalchemy.ext.asyncio import AsyncSession
from datetime import datetime, timedelta
from typing import List, Optional
import pandas as pd

class StockCRUD:
    """股票資料 CRUD"""

    @staticmethod
    async def get_stock(
        db: AsyncSession,
        ticker: str
    ) -> Optional[Stock]:
        """獲取股票資訊"""
        stmt = select(Stock).where(Stock.ticker == ticker)
        result = await db.execute(stmt)
        return result.scalar_one_or_none()

    @staticmethod
    async def get_ohlc(
        db: AsyncSession,
        ticker: str,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        limit: Optional[int] = None
    ) -> pd.DataFrame:
        """
        獲取 OHLC 資料

        Returns:
            pandas DataFrame
        """
        # 建立查詢
        stmt = select(StockOHLC).where(StockOHLC.ticker == ticker)

        if start_date:
            stmt = stmt.where(StockOHLC.timestamp >= start_date)
        if end_date:
            stmt = stmt.where(StockOHLC.timestamp <= end_date)

        stmt = stmt.order_by(StockOHLC.timestamp.desc())

        if limit:
            stmt = stmt.limit(limit)

        # 執行查詢
        result = await db.execute(stmt)
        rows = result.fetchall()

        # 轉換為 DataFrame
        if not rows:
            return pd.DataFrame()

        df = pd.DataFrame([
            {
                "date": row.timestamp,
                "open": float(row.open),
                "high": float(row.high),
                "low": float(row.low),
                "close": float(row.close),
                "volume": int(row.volume)
            }
            for row in rows
        ])

        return df.sort_values("date")

    @staticmethod
    async def get_latest_indicators(
        db: AsyncSession,
        ticker: str
    ) -> Optional[TechnicalIndicator]:
        """獲取最新技術指標"""
        stmt = (
            select(TechnicalIndicator)
            .where(TechnicalIndicator.ticker == ticker)
            .order_by(TechnicalIndicator.timestamp.desc())
            .limit(1)
        )

        result = await db.execute(stmt)
        return result.scalar_one_or_none()

    @staticmethod
    async def get_indicators_series(
        db: AsyncSession,
        ticker: str,
        days: int = 90
    ) -> pd.DataFrame:
        """獲取指標時間序列"""
        start_date = datetime.now() - timedelta(days=days)

        stmt = (
            select(TechnicalIndicator)
            .where(
                and_(
                    TechnicalIndicator.ticker == ticker,
                    TechnicalIndicator.timestamp >= start_date
                )
            )
            .order_by(TechnicalIndicator.timestamp)
        )

        result = await db.execute(stmt)
        rows = result.fetchall()

        if not rows:
            return pd.DataFrame()

        # 轉換為 DataFrame（簡化版）
        data = []
        for row in rows:
            data.append({
                "date": row.timestamp,
                "rsi": float(row.rsi) if row.rsi else None,
                "macd": float(row.macd) if row.macd else None,
                # ... 其他指標
            })

        return pd.DataFrame(data)

class ExecutionCRUD:
    """Agent 執行記錄 CRUD"""

    @staticmethod
    async def create_execution(
        db: AsyncSession,
        agent_name: str,
        task_type: str,
        task_input: dict,
        strategy: Optional[dict] = None,
        result: Optional[dict] = None,
        verification_passed: Optional[bool] = None,
        iterations: int = 0,
        execution_time_ms: Optional[int] = None,
        error: Optional[str] = None
    ) -> AgentExecution:
        """建立執行記錄"""
        execution = AgentExecution(
            agent_name=agent_name,
            task_type=task_type,
            task_input=task_input,
            strategy=strategy,
            result=result,
            verification_passed=verification_passed,
            iterations=iterations,
            execution_time_ms=execution_time_ms,
            error=error
        )

        db.add(execution)
        await db.commit()
        await db.refresh(execution)

        return execution

    @staticmethod
    async def get_failed_executions(
        db: AsyncSession,
        agent_name: str,
        days: int = 7,
        limit: int = 100
    ) -> List[AgentExecution]:
        """獲取失敗的執行記錄（用於優化策略）"""
        start_date = datetime.now() - timedelta(days=days)

        stmt = (
            select(AgentExecution)
            .where(
                and_(
                    AgentExecution.agent_name == agent_name,
                    AgentExecution.verification_passed == False,
                    AgentExecution.created_at >= start_date
                )
            )
            .order_by(AgentExecution.created_at.desc())
            .limit(limit)
        )

        result = await db.execute(stmt)
        return result.scalars().all()

Alembic 遷移

alembic.ini

[alembic]
script_location = alembic
sqlalchemy.url = postgresql+asyncpg://localhost/cortex_investment

[loggers]
keys = root,sqlalchemy,alembic

[handlers]
keys = console

[formatters]
keys = generic

建立遷移

# 初始化
alembic init alembic

# 建立遷移
alembic revision --autogenerate -m "Initial schema"

# 執行遷移
alembic upgrade head

資料庫配置

環境變數 (.env):

DATABASE_URL=postgresql+asyncpg://cortex:password@localhost:5432/cortex_investment
DB_POOL_SIZE=10
DB_MAX_OVERFLOW=20

連接範例:

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    os.getenv("DATABASE_URL"),
    echo=False,
    pool_size=int(os.getenv("DB_POOL_SIZE", 10)),
    max_overflow=int(os.getenv("DB_MAX_OVERFLOW", 20))
)

效能優化

索引策略

ticker + timestamp 複合索引（最常見查詢）
timestamp DESC 索引（時序查詢）
verification_passed 索引（失敗分析）

查詢優化

使用 LIMIT 限制結果數量
避免 SELECT *，只查詢需要的欄位
使用 prepared statements（SQLAlchemy 自動處理）

資料維護

-- 清理舊的執行記錄（保留 30 天）
DELETE FROM agent_executions
WHERE created_at < CURRENT_DATE - INTERVAL '30 days';

-- 更新統計資訊
ANALYZE stocks;
ANALYZE stock_ohlc;
ANALYZE technical_indicators;

文檔版本: 1.2 最後更新: 2026-07-31

Database Schema 文檔

概述

資料庫設計

ER Diagram

表結構定義

1. taiwan.stocks／us.stocks（股票基本資料）

2. taiwan.stock_ohlc／us.stock_ohlc（股票 OHLC 時序資料）

3. technical_indicators (技術指標)

4. agent_executions (Agent 執行記錄)

Token session monitoring (refresh_tokens)

Access-log 真實起訖時間 (api_access_log)

Theme recap 快取版本與 access-log 細部耗時

Theme store（題材正典）：public.theme / taiwan.theme_member

資料庫初始化

init_db.py

CRUD 操作

crud.py

Alembic 遷移

alembic.ini

建立遷移

資料庫配置

效能優化

索引策略

查詢優化

資料維護

1. `taiwan.stocks`／`us.stocks`（股票基本資料）

2. `taiwan.stock_ohlc`／`us.stock_ohlc`（股票 OHLC 時序資料）

Token session monitoring (`refresh_tokens`)

Access-log 真實起訖時間 (`api_access_log`)

Theme store（題材正典）：`public.theme` / `taiwan.theme_member`