""" process_daily_rates.py ======================= Processes the manually downloaded rate files into clean daily CSVs. Input files: UK BOE: data/raw/rates/glcnominalddata/ (8 Excel files, sheet "4. spot curve") Japan MoF: data/raw/rates/jgbc/jgbcme_all (CSV) Australia: data/raw/rates/RBA/ (Excel, sheet 0) Output files: data/raw/rates/uk2y_daily.csv data/raw/rates/uk10y_daily.csv data/raw/rates/jp2y_daily.csv data/raw/rates/jp10y_daily.csv data/raw/rates/au2y_daily.csv data/raw/rates/au10y_daily.csv Place this file in: C:\\Users\\paul_\\OneDrive\\fx_macro_intraday\\src\\ingestion\\process_daily_rates.py Run from project root: python src/ingestion/process_daily_rates.py """ import pandas as pd import numpy as np from pathlib import Path import openpyxl BASE_PATH = Path(__file__).resolve().parents[2] RATES_PATH = BASE_PATH / "data" / "raw" / "rates" BOE_DIR = RATES_PATH / "glcnominalddata" JGBC_DIR = RATES_PATH / "jgbc" RBA_DIR = RATES_PATH / "RBA" # ── Helpers ─────────────────────────────────────────────────────────────────── def save_series(df: pd.DataFrame, filename: str) -> None: out = RATES_PATH / filename df.to_csv(out, index=False) print(f" Saved {filename} | {len(df):,} rows | " f"{df['date'].min()} to {df['date'].max()}") def clean_and_save(dates, values, out_col: str, filename: str) -> pd.DataFrame: df = pd.DataFrame({"date": dates, out_col: values}) df["date"] = pd.to_datetime(df["date"], errors="coerce", dayfirst=True) df[out_col] = pd.to_numeric(df[out_col], errors="coerce") df = (df.dropna() .drop_duplicates(subset=["date"]) .sort_values("date") .reset_index(drop=True)) df["date"] = df["date"].dt.strftime("%Y-%m-%d") save_series(df, filename) return df # ── UK — Bank of England GLC Nominal ───────────────────────────────────────── def process_boe() -> tuple: """ BOE Excel files have 5 sheets: 'info', '1. fwds, short end', '2. fwd curve', '3. spot, short end', '4. spot curve' We need '4. spot curve' which contains daily spot yields at each maturity. Columns are maturity in years: 0.5, 1, 1.5, 2, 2.5, ... 25 First column is the date. """ print("\nProcessing UK BOE GLC Nominal files...") print(f" Source: {BOE_DIR}") excel_files = sorted(BOE_DIR.glob("*.xls")) + sorted(BOE_DIR.glob("*.xlsx")) if not excel_files: print(f" [ERROR] No Excel files found in {BOE_DIR}") return None, None print(f" Found {len(excel_files)} files") all_rows_2y = [] all_rows_10y = [] for fpath in excel_files: print(f"\n Reading {fpath.name}...") try: # Check available sheets wb = openpyxl.load_workbook(fpath, read_only=True) sheets = wb.sheetnames wb.close() # Find the spot curve sheet spot_sheet = None for s in sheets: if "spot" in s.lower() and "curve" in s.lower(): spot_sheet = s break if spot_sheet is None: spot_sheet = sheets[-1] # fallback to last sheet print(f" Sheets: {sheets}") print(f" Using: '{spot_sheet}'") # Read the sheet with no header to inspect structure df_raw = pd.read_excel(fpath, sheet_name=spot_sheet, header=None) print(f" Raw shape: {df_raw.shape}") # Find the header row — it contains numeric maturity values header_row_idx = 0 for i in range(min(15, len(df_raw))): row = [str(x) for x in df_raw.iloc[i].tolist()] row_str = " ".join(row).lower() if ("years" in row_str or any(s in row for s in ["0.5", "1.0", "2.0", "2", "10.0", "10"])): header_row_idx = i break df = pd.read_excel(fpath, sheet_name=spot_sheet, header=header_row_idx) df.columns = [str(c).strip() for c in df.columns] print(f" Header row: {header_row_idx}") print(f" Columns (first 12): {list(df.columns)[:12]}") print(f" Rows: {len(df)}") if len(df.columns) < 3: print(f" [WARNING] Too few columns — skipping") continue date_col = df.columns[0] col_2y = None col_10y = None for col in df.columns[1:]: try: val = float(str(col).replace("years", "").strip()) if abs(val - 2.0) < 0.1: col_2y = col if abs(val - 10.0) < 0.1: col_10y = col except (ValueError, TypeError): cs = str(col).lower().strip() if cs in ["2", "2.0", "2 yr", "2yr", "2 year"]: col_2y = col if cs in ["10", "10.0", "10 yr", "10yr", "10 year"]: col_10y = col print(f" 2Y: '{col_2y}' | 10Y: '{col_10y}'") if col_2y is None or col_10y is None: print(f" [WARNING] columns not found. All: {list(df.columns)}") continue dates = pd.to_datetime(df[date_col], errors="coerce") vals_2y = pd.to_numeric(df[col_2y], errors="coerce") vals_10y = pd.to_numeric(df[col_10y], errors="coerce") valid = dates.notna() & vals_2y.notna() & vals_10y.notna() print(f" Valid rows: {valid.sum()}") for date, v2, v10 in zip(dates[valid], vals_2y[valid], vals_10y[valid]): all_rows_2y.append((date, v2)) all_rows_10y.append((date, v10)) except Exception as e: print(f" [ERROR] {e}") import traceback traceback.print_exc() continue if not all_rows_2y: print(" [FAILED] No UK data extracted") return None, None dates_2y, vals_2y = zip(*all_rows_2y) dates_10y, vals_10y = zip(*all_rows_10y) uk2y = clean_and_save(dates_2y, vals_2y, "uk2y", "uk2y_daily.csv") uk10y = clean_and_save(dates_10y, vals_10y, "uk10y", "uk10y_daily.csv") return uk2y, uk10y # ── Japan — MoF JGB ─────────────────────────────────────────────────────────── def process_japan() -> tuple: """ MoF CSV: title row at 0, real headers (Date, 1Y, 2Y ... 10Y) at row 1. """ print("\nProcessing Japan MoF JGB file...") candidates = ( list(JGBC_DIR.glob("jgbcme_all*")) + list(JGBC_DIR.glob("jgbcm*all*")) + list(JGBC_DIR.glob("*.csv")) + list(JGBC_DIR.glob("*.xls*")) ) if not candidates: print(f" [ERROR] No files found in {JGBC_DIR}") return None, None fpath = candidates[0] print(f" Reading: {fpath.name}") try: df = None for skiprows in [1, 0]: try: if fpath.suffix.lower() == ".csv": df_try = pd.read_csv(fpath, encoding="utf-8", skiprows=skiprows, on_bad_lines="skip") elif fpath.suffix.lower() in [".xls", ".xlsx"]: df_try = pd.read_excel(fpath, skiprows=skiprows) else: try: df_try = pd.read_csv(fpath, encoding="utf-8", skiprows=skiprows, on_bad_lines="skip") except Exception: df_try = pd.read_csv(fpath, encoding="shift-jis", skiprows=skiprows, on_bad_lines="skip") df_try.columns = [str(c).strip() for c in df_try.columns] first_col = df_try.columns[0].lower() if "date" in first_col or "unnamed" not in first_col: df = df_try print(f" Using skiprows={skiprows}") break except Exception as e: print(f" skiprows={skiprows} failed: {e}") if df is None: print(" [ERROR] Could not read file") return None, None print(f" Columns: {list(df.columns)[:10]}") print(f" Rows: {len(df)}") date_col = df.columns[0] jp2y_col = None jp10y_col = None for col in df.columns: cs = str(col).lower().strip() if cs in ["2y", "2year", "2 year", "2yr", "2-year"]: jp2y_col = col if cs in ["10y", "10year", "10 year", "10yr", "10-year"]: jp10y_col = col if jp2y_col is None or jp10y_col is None: for col in df.columns: try: val = float( str(col).replace("year", "").replace("yr", "") .replace("y", "").strip() ) if abs(val - 2.0) < 0.1 and jp2y_col is None: jp2y_col = col if abs(val - 10.0) < 0.1 and jp10y_col is None: jp10y_col = col except (ValueError, TypeError): pass print(f" 2Y: '{jp2y_col}' | 10Y: '{jp10y_col}'") if jp2y_col is None or jp10y_col is None: print(f" [ERROR] columns not found. All: {list(df.columns)}") return None, None jp2y = clean_and_save(df[date_col], df[jp2y_col], "jp2y", "jp2y_daily.csv") jp10y = clean_and_save(df[date_col], df[jp10y_col], "jp10y", "jp10y_daily.csv") return jp2y, jp10y except Exception as e: print(f" [ERROR] {e}") import traceback traceback.print_exc() return None, None # ── Australia — RBA Table F2 ───────────────────────────────────────────────── def process_rba() -> tuple: """ RBA F2 Excel columns use series IDs not readable names: Series ID | FCMYGBAG2D | FCMYGBAG3D | FCMYGBAG5D | FCMYGBAG10D | FCMYGBAGID The '2D' suffix = 2-year, '10D' = 10-year. """ print("\nProcessing Australia RBA Table F2 file...") candidates = ( list(RBA_DIR.glob("f02d*")) + list(RBA_DIR.glob("f2*")) + list(RBA_DIR.glob("*.xls*")) + list(RBA_DIR.glob("*.csv")) ) if not candidates: print(f" [ERROR] No files found in {RBA_DIR}") return None, None fpath = candidates[0] print(f" Reading: {fpath.name}") try: df = None for skiprows in [10, 9, 11, 1, 0]: try: df_try = pd.read_excel(fpath, sheet_name=0, skiprows=skiprows) df_try.columns = [str(c).strip() for c in df_try.columns] first_col_sample = df_try.iloc[:3, 0].astype(str).tolist() has_dates = any( len(s) > 4 and any(c.isdigit() for c in s) for s in first_col_sample ) non_unnamed = sum( 1 for c in df_try.columns if "unnamed" not in c.lower() ) if has_dates and non_unnamed >= 2: df = df_try print(f" Using skiprows={skiprows}") print(f" Columns: {list(df.columns)}") break except Exception: continue if df is None: print(" [ERROR] Could not read RBA file") return None, None print(f" Rows: {len(df)}") date_col = df.columns[0] au2y_col = None au10y_col = None # RBA uses series codes: FCMYGBAG2D = 2-year, FCMYGBAG10D = 10-year for col in df.columns: cu = str(col).upper().strip() cs = str(col).lower().strip() # Match series code ending in 2D = 2-year if (cu.endswith("2D") and "FCMY" in cu) or \ any(x in cs for x in ["2-year", "2 year", "2year", "2yr"]): au2y_col = col # Match series code ending in 10D = 10-year if (cu.endswith("10D") and "FCMY" in cu) or \ any(x in cs for x in ["10-year", "10 year", "10year", "10yr"]): au10y_col = col print(f" 2Y: '{au2y_col}' | 10Y: '{au10y_col}'") if au2y_col is None or au10y_col is None: print(" [WARNING] Column names not matched. All columns:") for i, c in enumerate(df.columns): print(f" [{i}] '{c}'") # Positional fallback: date, 2Y, 3Y, 5Y, 10Y if len(df.columns) >= 5: au2y_col = df.columns[1] au10y_col = df.columns[4] elif len(df.columns) >= 3: au2y_col = df.columns[1] au10y_col = df.columns[2] print(f" Using positional: 2Y='{au2y_col}', 10Y='{au10y_col}'") au2y = clean_and_save(df[date_col], df[au2y_col], "au2y", "au2y_daily.csv") au10y = clean_and_save(df[date_col], df[au10y_col], "au10y", "au10y_daily.csv") return au2y, au10y except Exception as e: print(f" [ERROR] {e}") import traceback traceback.print_exc() return None, None # ── Main ────────────────────────────────────────────────────────────────────── def main(): print("=" * 65) print("PROCESS DAILY RATES") print(f" Reading from: {RATES_PATH}") print("=" * 65) results = {} print("\n" + "-" * 65) print("UK — Bank of England GLC Nominal") uk2y, uk10y = process_boe() results["uk2y"] = uk2y is not None and len(uk2y) > 100 results["uk10y"] = uk10y is not None and len(uk10y) > 100 print("\n" + "-" * 65) print("Japan — Ministry of Finance JGB") jp2y, jp10y = process_japan() results["jp2y"] = jp2y is not None and len(jp2y) > 100 results["jp10y"] = jp10y is not None and len(jp10y) > 100 print("\n" + "-" * 65) print("Australia — RBA Table F2") au2y, au10y = process_rba() results["au2y"] = au2y is not None and len(au2y) > 100 results["au10y"] = au10y is not None and len(au10y) > 100 print(f"\n{'=' * 65}") print("SUMMARY") print(f"{'=' * 65}") passed = sum(results.values()) print(f"\n {'Series':<16} {'Status':<10} {'File'}") print(f" {'-' * 45}") for name, ok in results.items(): status = "OK" if ok else "FAILED" print(f" {name:<16} {status:<10} {name}_daily.csv") print(f"\n Passed: {passed}/{len(results)}") if passed >= 4: print(""" Next steps: 1. Update PAIR_CONFIGS in multi_pair_signals_v1.py: GBPUSD for_2y : ("uk2y_daily.csv", "uk2y"), monthly: False GBPUSD for_10y: ("uk10y_daily.csv", "uk10y"), monthly: False USDJPY for_2y : ("jp2y_daily.csv", "jp2y"), monthly: False USDJPY for_10y: ("jp10y_daily.csv", "jp10y"), monthly: False AUDUSD for_2y : ("au2y_daily.csv", "au2y"), monthly: False AUDUSD for_10y: ("au10y_daily.csv", "au10y"), monthly: False 2. Run: python src/research/multi_pair_signals_v1.py """) else: failed = [k for k, v in results.items() if not v] print(f"\n Failed: {failed}") if __name__ == "__main__": main()