SQL×Pythonの“接着力”が実務を変える：粒度・主キー・時点→意思決定までの4ステップ

テーブル: orders
粒度: 注文行（order_id, line_no）
主キー: (order_id, line_no)
時刻: ordered_at（JST）
外部参照: customers(customer_id), price_history(sku, valid_from, valid_to)

WITH daily AS (
  SELECT DATE(ordered_at) AS d, SUM(amount) AS sales
  FROM order_lines
  GROUP BY DATE(ordered_at)
)
SELECT d, sales FROM daily ORDER BY d;
-- 粒度: 日

SELECT customer_id,
       MAX(ordered_at) OVER w AS last_order_at,
       COUNT(*)        OVER w AS freq,
       SUM(amount)     OVER w AS monetary
FROM orders
WINDOW w AS (PARTITION BY customer_id);
-- 粒度: 注文行（集計は窓で顧客粒度に投影）

SELECT o.order_id, o.sku, p.price
FROM orders o
JOIN price_history p
  ON o.sku = p.sku
 AND o.ordered_at >= p.valid_from
 AND (p.valid_to IS NULL OR o.ordered_at < p.valid_to);
-- 粒度: 注文行（“時点一致”での参照）

SELECT c.customer_id, a.attr, a.valid_from, a.valid_to
FROM customers c
JOIN customer_attr a
  ON c.customer_id = a.customer_id;
-- 下流JOINはP3のBetween条件で

SELECT * FROM customers c
LEFT JOIN churned_list x ON c.customer_id = x.customer_id
WHERE x.customer_id IS NULL;

SELECT * FROM (
  SELECT *, ROW_NUMBER() OVER(
           PARTITION BY order_id ORDER BY updated_at DESC) AS rn
  FROM orders_staging
) t WHERE rn = 1;

SELECT sku, COUNT(DISTINCT price) AS n
FROM price_history GROUP BY sku HAVING n > 1;
-- “同時点に2価格”の異常などを別途検査

SELECT user_id, session_id,
       MIN(ts) AS start_ts, MAX(ts) AS end_ts
FROM (
  SELECT *, SUM(new_session) OVER (
    PARTITION BY user_id ORDER BY ts) AS session_id
  FROM (
    SELECT *, CASE WHEN LAG(ts) OVER(PARTITION BY user_id ORDER BY ts)
                     < ts - INTERVAL '30 MINUTE' THEN 1 ELSE 0 END AS new_session
    FROM pageviews
  ) x
) y
GROUP BY user_id, session_id;

WITH spans AS (
  SELECT sku, valid_from, COALESCE(valid_to, CURRENT_DATE) AS valid_to, stock
  FROM stock_span
), days AS (
  SELECT sku, DATE(d) AS d, stock
  FROM spans, LATERAL (SELECT generate_series(valid_from, valid_to, INTERVAL '1 day') AS d) g
)
SELECT * FROM days;
-- 粒度: sku-日（倉庫の可用性を日次へ）

WITH base AS (
  SELECT DISTINCT o.order_id, l.sku
  FROM orders o JOIN order_lines l USING(order_id)
)
SELECT sku, COUNT(*) AS orders FROM base GROUP BY sku;
-- DISTINCTで粒度を“注文-商品”に正規化して計数

SELECT "group", AVG(age) AS avg_age, AVG(spend) AS avg_spend
FROM ab_users GROUP BY "group";
-- Python側でCI/検定へ

import sqlite3, pandas as pd
con = sqlite3.connect("app.db")
q = open("sql/feature_store.sql").read()
dtypes = {"customer_id": "Int64", "sku": "string"}
df = pd.read_sql_query(q, con, parse_dates=["ordered_at"], dtype=dtypes)
# 粒度の確認（主キーの一意性）
assert df.groupby(["order_id","line_no"]).size().le(1).all()

# タイムゾーン（DBはUTC想定）

# df\["ordered\_at"] = (df\["ordered\_at"]

# .dt.tz\_localize("UTC").dt.tz\_convert("Asia/Tokyo"))

# tests/test_feature_store.py
import pandas as pd

def test\_primary\_key\_unique(feature\_store: pd.DataFrame):
assert not feature\_store.duplicated(\["order\_id","line\_no"]).any()

def test\_value\_ranges(feature\_store):
assert feature\_store\["amount"].ge(0).all()

# 期間JOINの欠損比率なども検査推奨

project-sql-python/
  sql/feature_store.sql
  notebooks/10_eda.ipynb
  src/report.py
  tests/test_feature_store.py
  requirements.txt
  Dockerfile
  README.md  # 再現手順1画面
  dist/report_YYYYMMDD.pdf

再現手順：
1) docker build -t report .
2) docker run -v $(pwd):/work report python -m src.report
データ: /data/orders.csv, price_history.csv
出力: dist/report_YYYYMMDD.pdf