From da64544d3a05157b6df13c3266e9af475d7d2bcc Mon Sep 17 00:00:00 2001 From: Brent Neale Date: Mon, 16 Feb 2026 15:34:13 +1000 Subject: [PATCH] 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 --- sql/create_fx_candles.sql | 20 ++++++++++++++++++++ src/data_engine.py | 3 +++ src/get_candles.py | 1 + src/supabase_upload.py | 11 ++++++----- 4 files changed, 30 insertions(+), 5 deletions(-) create mode 100644 sql/create_fx_candles.sql diff --git a/sql/create_fx_candles.sql b/sql/create_fx_candles.sql new file mode 100644 index 0000000..86e0604 --- /dev/null +++ b/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) +); diff --git a/src/data_engine.py b/src/data_engine.py index 863aab8..d897d03 100644 --- a/src/data_engine.py +++ b/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 diff --git a/src/get_candles.py b/src/get_candles.py index d57ed0a..95588f8 100644 --- a/src/get_candles.py +++ b/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()) diff --git a/src/supabase_upload.py b/src/supabase_upload.py index d32b66b..61a3919 100644 --- a/src/supabase_upload.py +++ b/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.")