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. stocks (股票基本資料)
用途: 儲存股票基本資訊
欄位:
CREATE TABLE stocks (
ticker VARCHAR(20) PRIMARY KEY,
name VARCHAR(200),
market VARCHAR(20), -- taiwan, us, japan
industry VARCHAR(100),
sector VARCHAR(100),
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
);
CREATE INDEX idx_stocks_market ON stocks(market);
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 Stock(Base):
__tablename__ = "stocks"
ticker = Column(String(20), primary_key=True)
name = Column(String(200))
market = Column(String(20), nullable=False, index=True)
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 stocks (ticker, name, market, industry, sector) VALUES
('2330.TW', '台積電', 'taiwan', '半導體', '科技'),
('AAPL', 'Apple Inc.', 'us', '消費性電子', '科技'),
('TSLA', 'Tesla Inc.', 'us', '電動車', '汽車');
2. stock_ohlc (股票 OHLC 時序資料)
用途: 儲存每日 OHLC (Open, High, Low, Close) 和成交量
欄位:
CREATE TABLE stock_ohlc (
ticker VARCHAR(20) NOT NULL,
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 (ticker, timestamp),
FOREIGN KEY (ticker) REFERENCES stocks(ticker) ON DELETE CASCADE
);
-- 索引（加速查詢）
CREATE INDEX idx_ohlc_ticker_time ON stock_ohlc(ticker, timestamp DESC);
CREATE INDEX idx_ohlc_timestamp ON stock_ohlc(timestamp DESC);
Python Model:
from sqlalchemy import Column, String, DateTime, Numeric, BigInteger, ForeignKey
class StockOHLC(Base):
__tablename__ = "stock_ohlc"
ticker = Column(String(20), ForeignKey("stocks.ticker"), primary_key=True)
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 stock_ohlc
WHERE ticker = '2330.TW'
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 (ticker, timestamp),
FOREIGN KEY (ticker) REFERENCES stocks(ticker) ON DELETE CASCADE
);
-- 索引
CREATE INDEX idx_indicators_ticker_time ON technical_indicators(ticker, timestamp DESC);
Python Model:
class TechnicalIndicator(Base):
__tablename__ = "technical_indicators"
ticker = Column(String(20), ForeignKey("stocks.ticker"), 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;
資料庫初始化
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.0
最後更新: 2025-02-01