Add granularity tracking, drop intermediate columns, add SQL schema

Track granularity (M1/M5) through the pipeline so rows are
distinguishable after upload. Drop helper columns (tr, typical_price,
pv) from the feature DataFrame. Add SQL schema with composite PK on
(time, instrument, granularity).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

sql/create_fx_candles.sql

@@ -0,0 +1,20 @@
+CREATE TABLE fx_candles (
+    time         TIMESTAMPTZ   NOT NULL,
+    instrument   TEXT          NOT NULL,
+    granularity  TEXT          NOT NULL,
+    open         DOUBLE PRECISION,
+    high         DOUBLE PRECISION,
+    low          DOUBLE PRECISION,
+    close        DOUBLE PRECISION,
+    volume       INTEGER,
+    ret          DOUBLE PRECISION,
+    logret       DOUBLE PRECISION,
+    sma_3        DOUBLE PRECISION,
+    sma_20       DOUBLE PRECISION,
+    ema_20       DOUBLE PRECISION,
+    rsi_14       DOUBLE PRECISION,
+    vol_20       DOUBLE PRECISION,
+    atr_14       DOUBLE PRECISION,
+    vwap_20      DOUBLE PRECISION,
+    PRIMARY KEY (time, instrument, granularity)
+);

src/data_engine.py

@@ -124,4 +124,7 @@ def build_all_features(df, config=None):
     df = add_atr(df, period=atr_p)
     df = add_vwap(df, period=vwap_w)
 
+    # Drop intermediate helper columns
+    df.drop(columns=["pv", "typical_price", "tr"], inplace=True, errors="ignore")
+
     return df

src/get_candles.py

@@ -49,6 +49,7 @@ def main():
             )
             df = candles_to_df(candles)
             df = build_all_features(df, config=feature_cfg)
+            df["granularity"] = granularity
             print(df.tail(10).to_string())
 
 

src/supabase_upload.py

@@ -23,7 +23,7 @@ if not SUPABASE_URL or not SUPABASE_KEY:
 supabase = create_client(SUPABASE_URL, SUPABASE_KEY)
 
 
-def df_to_records(df: pd.DataFrame, instrument: str):
+def df_to_records(df: pd.DataFrame, instrument: str, granularity: str = "M1"):
     """
     Convert DataFrame to a list of dicts suitable for Supabase/Postgres.
     Ensures time is ISO string and numeric types are regular Python types.
@@ -36,8 +36,9 @@ def df_to_records(df: pd.DataFrame, instrument: str):
         df2.index.name = "time"
     df2 = df2.reset_index()
 
-    # Add instrument column
+    # Add instrument and granularity columns
     df2["instrument"] = instrument
+    df2["granularity"] = granularity
 
     # Convert Timestamp -> ISO string (timezone-aware preserved)
     df2["time"] = df2["time"].apply(lambda t: pd.to_datetime(t).isoformat())
@@ -95,8 +96,8 @@ def df_to_records(df: pd.DataFrame, instrument: str):
     return records
 
 
-def upload_dataframe(df: pd.DataFrame, instrument="EUR_USD", chunk_size=500):
-    records = df_to_records(df, instrument)
+def upload_dataframe(df: pd.DataFrame, instrument="EUR_USD", granularity="M1", chunk_size=500):
+    records = df_to_records(df, instrument, granularity)
 
     # debug: check first record is JSON serializable
     if records:
@@ -165,6 +166,6 @@ if __name__ == "__main__":
             df = candles_to_df(data)
             df = build_all_features(df, config=feature_cfg)
 
-            upload_dataframe(df, instrument=instrument, chunk_size=200)
+            upload_dataframe(df, instrument=instrument, granularity=granularity, chunk_size=200)
 
     print("Upload completed.")