PostgreSQL + TimescaleDBで時系列データを管理する：個人開発のDB設計

1
09:00 ジョブA: メトリクスデータ書き込み
2
09:00 ジョブB: 分析結果書き込み（ジョブAと同時実行）
3
09:00 ジョブC: パフォーマンス読み取り（↑と同時実行）
4
→ SQLite: "database is locked" エラー

1
CREATE TABLE daily_metrics (
2
    id BIGSERIAL PRIMARY KEY,
3
    datetime TIMESTAMPTZ NOT NULL,
4
    source VARCHAR(20) NOT NULL,
5
    timeframe VARCHAR(10) NOT NULL,
6
    value DOUBLE PRECISION NOT NULL,
7
    min_value DOUBLE PRECISION NOT NULL,
8
    max_value DOUBLE PRECISION NOT NULL,
9
    avg_value DOUBLE PRECISION NOT NULL,
10
    sample_count DOUBLE PRECISION,
11
    created_at TIMESTAMPTZ DEFAULT NOW(),
12
    UNIQUE (datetime, source, timeframe)
13
);
14

15
-- インデックス
2 collapsed lines
16
CREATE INDEX idx_metrics_source_time
17
    ON daily_metrics (source, datetime DESC);

1
CREATE TABLE analysis_results (
2
    id BIGSERIAL PRIMARY KEY,
3
    datetime TIMESTAMPTZ NOT NULL,
4
    source VARCHAR(20) NOT NULL,
5
    timeframe VARCHAR(10) NOT NULL,
6
    metric_name VARCHAR(50) NOT NULL,
7
    metric_value DOUBLE PRECISION,
8
    reference_date DATE NOT NULL,
9
    composite_id VARCHAR(100) NOT NULL,
10
    created_at TIMESTAMPTZ DEFAULT NOW()
11
);
12

13
-- BIツール対応: 複合IDインデックス
14
CREATE INDEX idx_analysis_composite
15
    ON analysis_results (composite_id);

1
CREATE TABLE performance_snapshots (
2
    datetime TIMESTAMPTZ NOT NULL,
3
    metric_name VARCHAR(50) NOT NULL,
4
    metric_value DOUBLE PRECISION NOT NULL,
5
    source VARCHAR(20),
6
    timeframe VARCHAR(10),
7
    metadata JSONB
8
);
9

10
-- TimescaleDB hypertable に変換（自動パーティション）
11
SELECT create_hypertable(
12
    'performance_snapshots',
13
    'datetime',
14
    chunk_time_interval => INTERVAL '1 month'
15
);

1
from myModules.mgmtDB import get_db_connection
2

3

4
def save_analysis_results(df: pd.DataFrame, table_name: str) -> None:
5
    """分析結果をDBに保存する。
6

7
    Args:
8
        df: 保存するDataFrame
9
        table_name: 保存先テーブル名
10
    """
11
    with get_db_connection() as conn:
12
        df.to_sql(table_name, conn, if_exists="append", index=False)

1
import pandas as pd
2
from myModules.mgmtDB import get_db_connection
3

4

5
def get_recent_performance(
6
    days: int = 30,
7
    source: str | None = None,
8
) -> pd.DataFrame:
9
    """直近N日のパフォーマンスデータを取得する。"""
10
    query = """
11
        SELECT datetime, metric_name, metric_value, source
12
        FROM performance_snapshots
13
        WHERE datetime >= NOW() - INTERVAL '%s days'
14
    """
15
    params = [days]
9 collapsed lines
16

17
    if source is not None:
18
        query += " AND source = %s"
19
        params.append(source)
20

21
    query += " ORDER BY datetime DESC"
22

23
    with get_db_connection() as conn:
24
        return pd.read_sql(query, conn, params=params)

1
-- 月単位の自動パーティション
2
SELECT create_hypertable(
3
    'performance_snapshots',
4
    'datetime',
5
    chunk_time_interval => INTERVAL '1 month'
6
);
7

8
-- 古いデータの一括削除（パーティション単位で高速）
9
SELECT drop_chunks('performance_snapshots', older_than => INTERVAL '1 year');

1
-- 直近30日の日次平均を取得
2
SELECT time_bucket('1 day', datetime) AS day,
3
       avg(metric_value) AS avg_value
4
FROM performance_snapshots
5
WHERE metric_name = 'daily_throughput'
6
  AND datetime >= NOW() - INTERVAL '30 days'
7
GROUP BY day
8
ORDER BY day;

1
from datetime import datetime
2

3

4
def generate_composite_id(
5
    source: str,
6
    timeframe: str,
7
    sequence: int,
8
) -> str:
9
    """複合IDを生成する。
10

11
    形式: {YYYYMMDD}_{Source}_{TimeFrame}_{YYYYMMDDHHMM}_{00001-99999}
12
    例: 20260327_server01_4H_202603271500_00001
13
    """
14
    today = datetime.now().strftime("%Y%m%d")
15
    now = datetime.now().strftime("%Y%m%d%H%M")
1 collapsed line
16
    return f"{today}_{source}_{timeframe}_{now}_{sequence:05d}"