Retail turn around

"""

Delta PSR ops pipeline

======================

What this does

--------------

1. Builds facility-day features from daily trade data + facility parquets + PSR.

2. Builds day-over-day delta dataset.

3. Trains a model on historical delta_psr_log over a user-selected training window.

4. Scores an out-of-sample evaluation window starting from a user-selected date.

5. Creates:

- global drivers

- facility-level drivers

- ops worklist for the latest evaluated day

6. Exports outputs to Excel.

Key design choices

------------------

- Model target = delta_psr_log

- Ops prioritisation uses:

* abs(delta_psr)

* shock_score

- Explanations use XGBoost pred_contribs

User controls

-------------

- TRAIN_START_DATE: first date to consider for training history

- TRAIN_NUM_DAYS: number of business dates from TRAIN_START_DATE

- EVAL_START_DATE: first date for out-of-sample evaluation

- weekends are ignored automatically

- dates with missing parquet files are skipped automatically

"""

from future import annotations

import glob

from pathlib import Path

from typing import Dict, List, Optional, Tuple

import numpy as np

import pandas as pd

import xgboost as xgb

from sklearn.metrics import mean_absolute_error, r2_score

# =========================================================

# USER PARAMETERS

# =========================================================

BASE_PATH = Path("impact")

OUTPUT_XLSX = Path("delta_psr_ops_output.xlsx")

TRAIN_START_DATE = "2025-10-03" # inclusive

TRAIN_NUM_DAYS = 30 # business/file dates from start

EVAL_START_DATE = "2025-11-20" # first out-of-sample after_date

ID_COL = "facility_id"

DATE_COL = "cob_date"

TARGET_COL = "psr"

TARGET_LOG_COL = "psr_log"

TARGET_DELTA_COL = "delta_psr"

TARGET_DELTA_LOG_COL = "delta_psr_log"

MIN_HISTORY_FOR_SHOCK = 5

XGB_PARAMS = {

"n_estimators": 600,

"max_depth": 6,

"learning_rate": 0.03,

"subsample": 0.8,

"colsample_bytree": 0.8,

"reg_alpha": 0.0,

"reg_lambda": 1.0,

"random_state": 42,

"n_jobs": -1,

}

TRADE_DUMMY_COLS = [

"agreementType_request", "buySell", "ccpGfcid", "cleared_ind", "clearing_type",

"cpnFlag", "csaMarginType", "dsft_eligible_flag", "fullTwoWay", "haircutType",

"illiquid_flag", "isCds", "isCdsIndex", "isEquityIndexBasket",

"isEquityIndexBasketFromSingleName", "isOption", "margin_eligible",

"matchType", "mrnccdIndicator", "multiBranchIndicator", "netting_type",

"option_early_termination_ind", "putCallInd", "regImMonitoringFlag",

"securityIDType", "settlementIndicator", "special_crf_ind", "usePsrMargin"

]

TRADE_NUMERIC_AGGS = {

"trade_count": ("transaction_id", "count"),

"amount_sum": ("amount", "sum"),

"amount_mean": ("amount", "mean"),

"amount_std": ("amount", "std"),

"amount_min": ("amount", "min"),

"amount_max": ("amount", "max"),

"days_to_maturity_mean": ("days_to_maturity", "mean"),

"days_to_maturity_max": ("days_to_maturity", "max"),

}

pd.options.mode.chained_assignment = None

# =========================================================

# REQUIRED: keep your existing downcast(df)

# =========================================================

# def downcast(df):

# ...

# =========================================================

# DATE HELPERS

# =========================================================

def business_dates_from(start_date: str, num_days: int) -> List[str]:

"""

Return business days (Mon-Fri) in YYYYMMDD format, starting from start_date,

keeping exactly num_days weekdays. Missing parquet dates are filtered later.

"""

dates = pd.bdate_range(start=start_date, periods=num_days)

return [d.strftime("%Y%m%d") for d in dates]

def all_business_dates_from(start_date: str, end_date: Optional[str] = None) -> List[str]:

"""

Return business days from start_date to end_date inclusive.

If end_date is None, generate up to today.

"""

if end_date is None:

end_date = pd.Timestamp.today().strftime("%Y-%m-%d")

dates = pd.bdate_range(start=start_date, end=end_date)

return [d.strftime("%Y%m%d") for d in dates]

def date_key_to_ts(x) -> pd.Timestamp:

return pd.to_datetime(str(x), format="%Y%m%d")

def ts_to_date_key(x: pd.Timestamp) -> str:

return pd.Timestamp(x).strftime("%Y%m%d")

# =========================================================

# FILE HELPERS

# =========================================================

def first_matching_file(pattern: str) -> Optional[str]:

matches = glob.glob(pattern)

return matches[0] if matches else None

def has_required_files(cob_date: str, base_path: Path = BASE_PATH) -> bool:

"""

Check whether the date has the required parquet inputs.

Weekends / missing dates are naturally skipped because files won't exist.

"""

parquet_dir = base_path / cob_date / "parquet"

psr_file = first_matching_file(str(parquet_dir / "*psr_breakdown*.parquet"))

margins_file = parquet_dir / "margins_by_facility.parquet"

mpor_file = parquet_dir / "mpor_by_facility.parquet"

coll_file = parquet_dir / "collaterals_by_facility.parquet"

trades_file = parquet_dir / "trades_with_tracing_logs.parquet"

return (

psr_file is not None

and trades_file.exists()

and margins_file.exists()

and mpor_file.exists()

and coll_file.exists()

)

def get_available_dates(

start_date: str,

num_days: Optional[int] = None,

end_date: Optional[str] = None,

base_path: Path = BASE_PATH,

) -> List[str]:

"""

Get available business dates with all required files.

"""

if num_days is not None:

raw_dates = business_dates_from(start_date, num_days)

else:

raw_dates = all_business_dates_from(start_date, end_date=end_date)

return [d for d in raw_dates if has_required_files(d, base_path)]

# =========================================================

# BASIC HELPERS

# =========================================================

def flatten_columns(df: pd.DataFrame) -> pd.DataFrame:

out = df.copy()

out.columns = [

"_".join([str(i) for i in col if str(i) != ""]).strip("_")

if isinstance(col, tuple)

else str(col)

for col in out.columns

]

return out

def p90(x: pd.Series) -> float:

return x.quantile(0.90)

def standardise_id(df: pd.DataFrame, id_col: str = ID_COL) -> pd.DataFrame:

out = df.copy()

out[id_col] = pd.to_numeric(out[id_col], errors="coerce")

out[id_col] = out[id_col].replace([np.inf, -np.inf], np.nan)

out = out[out[id_col].notna()].copy()

out[id_col] = out[id_col].astype("int64")

return out

def clip_inf_and_downcast(df: pd.DataFrame) -> pd.DataFrame:

out = df.replace([np.inf, -np.inf], np.nan)

return downcast(out)

# =========================================================

# LOAD RAW DAILY FILES

# =========================================================

def load_psr_for_date(cob_date: str, base_path: Path = BASE_PATH) -> pd.DataFrame:

"""

Load PSR breakdown, keep one row per facility_id using max PSR / MLIV logic.

Also create psr_log.

"""

pattern = str(base_path / cob_date / "parquet" / "*psr_breakdown*.parquet")

file_path = first_matching_file(pattern)

if file_path is None:

raise FileNotFoundError(f"No PSR breakdown file for {cob_date}")

df = pd.read_parquet(file_path)

df = standardise_id(df, ID_COL)

df[TARGET_COL] = pd.to_numeric(df[TARGET_COL], errors="coerce")

keep_cols = [c for c in [ID_COL, TARGET_COL, "mliv"] if c in df.columns]

out = df.groupby(ID_COL, as_index=False)[keep_cols[1:]].max()

out = out[out[TARGET_COL].fillna(0) != 0].copy()

out[TARGET_LOG_COL] = np.log1p(out[TARGET_COL].clip(lower=0))

out[DATE_COL] = date_key_to_ts(cob_date)

return clip_inf_and_downcast(out)

def load_trades_for_date(cob_date: str, base_path: Path = BASE_PATH) -> pd.DataFrame:

file_path = base_path / cob_date / "parquet" / "trades_with_tracing_logs.parquet"

df = pd.read_parquet(file_path)

df = standardise_id(df, ID_COL)

# minimal prep only

if "amount" in df.columns:

df["amount"] = pd.to_numeric(df["amount"], errors="coerce")

if "maturity_date" in df.columns:

df["maturity_date"] = pd.to_datetime(df["maturity_date"], format="%Y%m%d", errors="coerce")

if DATE_COL not in df.columns:

df[DATE_COL] = date_key_to_ts(cob_date)

else:

df[DATE_COL] = pd.to_datetime(df[DATE_COL], errors="coerce")

if "days_to_maturity" not in df.columns and {"maturity_date", DATE_COL}.issubset(df.columns):

df["days_to_maturity"] = np.busday_count(

df[DATE_COL].values.astype("datetime64[D]"),

df["maturity_date"].values.astype("datetime64[D]")

)

df[DATE_COL] = date_key_to_ts(cob_date)

return clip_inf_and_downcast(df)

def load_daily_inputs(

dates: List[str],

base_path: Path = BASE_PATH,

) -> Tuple[Dict[str, pd.DataFrame], Dict[str, pd.DataFrame]]:

"""

Load dfs and prs dictionaries for the supplied available dates only.

"""

dfs = {}

prs = {}

for d in dates:

print(f"Loading raw inputs: {d}")

prs[d] = load_psr_for_date(d, base_path)

dfs[d] = load_trades_for_date(d, base_path)

return dfs, prs

# =========================================================

# FACILITY-DAY FEATURE ENGINEERING

# =========================================================

def make_trade_numeric_agg(trade_df: pd.DataFrame) -> pd.DataFrame:

trade_df = standardise_id(trade_df, ID_COL).copy()

agg_map = {k: v for k, v in TRADE_NUMERIC_AGGS.items() if v[0] in trade_df.columns}

if "days_to_maturity" in trade_df.columns:

agg_map["days_to_maturity_p90"] = ("days_to_maturity", p90)

if not agg_map:

return trade_df[[ID_COL]].drop_duplicates().reset_index(drop=True)

out = trade_df.groupby(ID_COL, dropna=False).agg(**agg_map).reset_index()

return clip_inf_and_downcast(out)

def make_trade_dummy_counts(trade_df: pd.DataFrame, dummy_cols: List[str]) -> pd.DataFrame:

trade_df = standardise_id(trade_df, ID_COL).copy()

cols = [c for c in dummy_cols if c in trade_df.columns]

if not cols:

return trade_df[[ID_COL]].drop_duplicates().reset_index(drop=True)

dummies = pd.get_dummies(trade_df[cols], dummy_na=True)

dummies[ID_COL] = trade_df[ID_COL].values

out = dummies.groupby(ID_COL, dropna=False).sum().reset_index()

return clip_inf_and_downcast(out)

def make_maturity_bucket_features(trade_df: pd.DataFrame) -> pd.DataFrame:

trade_df = standardise_id(trade_df, ID_COL).copy()

if not {ID_COL, "amount", "days_to_maturity"}.issubset(trade_df.columns):

return trade_df[[ID_COL]].drop_duplicates().reset_index(drop=True)

x = trade_df[[ID_COL, "amount", "days_to_maturity"]].copy()

x["amount"] = pd.to_numeric(x["amount"], errors="coerce")

x["days_to_maturity"] = pd.to_numeric(x["days_to_maturity"], errors="coerce")

bins = [-np.inf, 0, 7, 35, 96, 187, 370, 1100, 1831, 2926, 3657, 5483, 10007, np.inf]

labels = [

"<0", "01_0_7", "02_8_35", "03_36_96", "04_97_187", "05_188_370",

"06_371_1100", "07_1101_1831", "08_1832_2926", "09_2927_3657",

"10_3658_5483", "11_5484_10007", ">10007"

]

x["maturity_bucket"] = pd.cut(x["days_to_maturity"], bins=bins, labels=labels)

grouped = (

x.groupby([ID_COL, "maturity_bucket"], observed=False)

.agg(

amount=("amount", "sum"),

days_to_maturity=("days_to_maturity", "mean"),

num_trx=("amount", "size"),

)

.reset_index()

)

grouped["amount_total"] = grouped.groupby(ID_COL)["amount"].transform("sum")

grouped["amount_pct"] = np.where(grouped["amount_total"] == 0, 0, grouped["amount"] / grouped["amount_total"])

pivoted = grouped.pivot(

index=ID_COL,

columns="maturity_bucket",

values=["amount", "days_to_maturity", "num_trx", "amount_pct"]

).fillna(0)

pivoted = flatten_columns(pivoted).reset_index()

return clip_inf_and_downcast(pivoted)

def safe_read_facility_parquet(cob_date: str, filename: str, base_path: Path = BASE_PATH) -> pd.DataFrame:

df = pd.read_parquet(base_path / cob_date / "parquet" / filename)

df = standardise_id(df, ID_COL)

return clip_inf_and_downcast(df)

def build_facility_day_dataset(

cob_date: str,

trade_df: pd.DataFrame,

psr_df: pd.DataFrame,

base_path: Path = BASE_PATH,

) -> pd.DataFrame:

"""

Build one row per facility_id for one date.

"""

num_agg = make_trade_numeric_agg(trade_df)

bucket_agg = make_maturity_bucket_features(trade_df)

dummy_agg = make_trade_dummy_counts(trade_df, TRADE_DUMMY_COLS)

margins = safe_read_facility_parquet(cob_date, "margins_by_facility.parquet", base_path)

mpor = safe_read_facility_parquet(cob_date, "mpor_by_facility.parquet", base_path)

coll = safe_read_facility_parquet(cob_date, "collaterals_by_facility.parquet", base_path)

psr_keep = [c for c in [ID_COL, TARGET_COL, TARGET_LOG_COL, "mliv"] if c in psr_df.columns]

psr_small = psr_df[psr_keep].drop_duplicates(subset=[ID_COL]).copy()

out = (

num_agg

.merge(bucket_agg, on=ID_COL, how="outer")

.merge(dummy_agg, on=ID_COL, how="outer")

.merge(margins, on=ID_COL, how="left")

.merge(mpor, on=ID_COL, how="left")

.merge(coll, on=ID_COL, how="left")

.merge(psr_small, on=ID_COL, how="outer")

)

out[DATE_COL] = date_key_to_ts(cob_date)

return clip_inf_and_downcast(out)

# =========================================================

# DELTA DATASET

# =========================================================

def make_delta_dataset(before_df: pd.DataFrame, after_df: pd.DataFrame) -> pd.DataFrame:

"""

Build day-over-day facility-level delta dataset.

"""

before_df = before_df.copy().set_index(ID_COL)

after_df = after_df.copy().set_index(ID_COL)

before_df, after_df = before_df.align(after_df, join="outer", axis=0)

b = before_df.drop(columns=[DATE_COL], errors="ignore")

a = after_df.drop(columns=[DATE_COL], errors="ignore")

is_new = b.isna().all(axis=1) & a.notna().any(axis=1)

is_removed = a.isna().all(axis=1) & b.notna().any(axis=1)

num_cols = sorted(set(b.select_dtypes(include=np.number).columns) | set(a.select_dtypes(include=np.number).columns))

b = b.reindex(columns=num_cols)

a = a.reindex(columns=num_cols)

delta = a.fillna(0) - b.fillna(0)

pct = delta / b.replace(0, np.nan)

out = pd.DataFrame(index=delta.index)

out["before_date"] = before_df[DATE_COL].dropna().iloc[0]

out["after_date"] = after_df[DATE_COL].dropna().iloc[0]

out["status"] = np.select([is_new, is_removed], ["new", "removed"], default="existing")

for c in num_cols:

out[c] = a[c]

out[f"delta_{c}"] = delta[c]

out[f"delta_pct_{c}"] = pct[c]

out = out.reset_index()

if f"delta_{TARGET_COL}" in out.columns:

out[TARGET_DELTA_COL] = out[f"delta_{TARGET_COL}"]

if f"delta_{TARGET_LOG_COL}" in out.columns:

out[TARGET_DELTA_LOG_COL] = out[f"delta_{TARGET_LOG_COL}"]

return clip_inf_and_downcast(out)

# =========================================================

# SHOCK SCORE

# =========================================================

def add_shock_score(delta_store: pd.DataFrame) -> pd.DataFrame:

"""

shock_score =

abs(delta_psr_log - facility_mean_delta_psr_log) / facility_std_delta_psr_log

"""

out = delta_store.copy()

hist = (

out.groupby(ID_COL)[TARGET_DELTA_LOG_COL]

.agg(["mean", "std", "count"])

.reset_index()

.rename(columns={

"mean": "hist_mean_delta_psr_log",

"std": "hist_std_delta_psr_log",

"count": "hist_n_obs",

})

)

out = out.merge(hist, on=ID_COL, how="left")

out["shock_score"] = (

(out[TARGET_DELTA_LOG_COL] - out["hist_mean_delta_psr_log"]).abs()

/ out["hist_std_delta_psr_log"].replace(0, np.nan)

)

out.loc[out["hist_n_obs"] < MIN_HISTORY_FOR_SHOCK, "shock_score"] = np.nan

out["shock_score"] = out["shock_score"].fillna(0).clip(upper=10)

return clip_inf_and_downcast(out)

# =========================================================

# TRAIN / EVAL SPLITS

# =========================================================

def get_train_eval_dates(

available_dates: List[str],

train_start_date: str,

train_num_days: int,

eval_start_date: str,

) -> Tuple[List[str], List[str]]:

"""

Build train/eval date sets using available file dates only.

Train dates:

first TRAIN_NUM_DAYS available dates from TRAIN_START_DATE

Eval dates:

all available dates from EVAL_START_DATE onwards

Notes:

- These are raw cob_dates for facility-day construction.

- Delta after_date values start from the second date in each sequence.

"""

train_start_key = pd.Timestamp(train_start_date).strftime("%Y%m%d")

eval_start_key = pd.Timestamp(eval_start_date).strftime("%Y%m%d")

available_dates = sorted(available_dates)

train_candidates = [d for d in available_dates if d >= train_start_key]

train_dates = train_candidates[:train_num_days]

eval_dates = [d for d in available_dates if d >= eval_start_key]

# include the previous available date before eval start so first eval delta can be formed

prev_eval_candidates = [d for d in available_dates if d < eval_start_key]

if prev_eval_candidates:

eval_dates = [prev_eval_candidates[-1]] + eval_dates

return train_dates, eval_dates

# =========================================================

# MODEL MATRIX

# =========================================================

def build_model_matrix(delta_store: pd.DataFrame):

"""

target = delta_psr_log

features = delta_* excluding leakage fields

"""

model_df = delta_store.copy()

model_df = model_df[(model_df["status"] == "existing") & (model_df[TARGET_DELTA_LOG_COL].notna())].copy()

exclude = {

ID_COL, "before_date", "after_date", "status",

TARGET_COL, TARGET_LOG_COL, "mliv",

TARGET_DELTA_COL, TARGET_DELTA_LOG_COL,

f"delta_pct_{TARGET_COL}", f"delta_pct_{TARGET_LOG_COL}",

"shock_score", "hist_mean_delta_psr_log", "hist_std_delta_psr_log", "hist_n_obs",

}

feature_cols = [c for c in model_df.columns if c.startswith("delta_") and c not in exclude]

X = model_df[feature_cols].replace([np.inf, -np.inf], np.nan).fillna(0).copy()

y = model_df[TARGET_DELTA_LOG_COL].astype(float).copy()

return model_df, X, y, feature_cols

# =========================================================

# MODEL

# =========================================================

def train_xgb_model(X: pd.DataFrame, y: pd.Series):

model = xgb.XGBRegressor(**XGB_PARAMS)

model.fit(X, y)

return model

def calculate_xgb_contributions(model, X_df: pd.DataFrame):

booster = model.get_booster()

dm = xgb.DMatrix(X_df, feature_names=X_df.columns.tolist())

contribs = booster.predict(dm, pred_contribs=True)

shap_df = pd.DataFrame(contribs, columns=X_df.columns.tolist() + ["bias"])

imp = shap_df[X_df.columns].abs().mean().sort_values(ascending=False)

imp = (100 * imp / imp.sum()).reset_index()

imp.columns = ["feature", "importance_pct"]

return imp, shap_df

# =========================================================

# FACILITY LEVEL DRIVERS / OPS REASONS

# =========================================================

def feature_to_reason(feature: str, value) -> str:

try:

v = float(value) if pd.notna(value) else np.nan

except Exception:

v = value

if "trade_count" in feature:

return f"Trade count changed by {v:,.0f}"

if "amount_sum" in feature:

return f"Total traded amount changed by {v:,.0f}"

if "amount_mean" in feature:

return f"Average trade amount changed by {v:,.0f}"

if "amount_pct" in feature:

return "Exposure mix shifted across maturity buckets"

if "days_to_maturity" in feature:

return "Maturity profile shifted"

if "vm" in feature.lower():

return f"Variation margin related metric changed by {v:,.0f}"

if "im" in feature.lower():

return f"Initial margin related metric changed by {v:,.0f}"

if "collateral" in feature.lower():

return f"Collateral related metric changed by {v:,.0f}"

if "num_trx" in feature:

return f"Number of trades in a maturity bucket changed by {v:,.0f}"

return f"{feature} changed by {v:,.2f}" if isinstance(v, (int, float, np.floating)) and pd.notna(v) else str(feature)

def get_facility_drivers(model_df: pd.DataFrame, X_df: pd.DataFrame, shap_df: pd.DataFrame, top_n: int = 10):

out = []

shap_only = shap_df[X_df.columns].reset_index(drop=True)

model_df = model_df.reset_index(drop=True)

X_df = X_df.reset_index(drop=True)

for i in range(len(model_df)):

tmp = pd.DataFrame({

"feature": X_df.columns,

"feature_delta": X_df.iloc[i].values,

"contribution": shap_only.iloc[i].values,

})

tmp["abs_contribution"] = tmp["contribution"].abs()

tmp = tmp.sort_values("abs_contribution", ascending=False).head(top_n)

tmp.insert(0, ID_COL, model_df.loc[i, ID_COL])

tmp.insert(1, "after_date", model_df.loc[i, "after_date"])

tmp.insert(2, TARGET_DELTA_COL, model_df.loc[i, TARGET_DELTA_COL])

tmp.insert(3, TARGET_DELTA_LOG_COL, model_df.loc[i, TARGET_DELTA_LOG_COL])

tmp["rank"] = np.arange(1, len(tmp) + 1)

out.append(tmp)

return clip_inf_and_downcast(pd.concat(out, ignore_index=True))

def build_ops_worklist(

scored_df: pd.DataFrame,

shap_df: pd.DataFrame,

feature_cols: List[str],

latest_after_date: pd.Timestamp,

top_n_reasons: int = 3,

) -> pd.DataFrame:

"""

Build latest-day ops queue.

"""

df = scored_df[scored_df["after_date"] == latest_after_date].copy().reset_index(drop=True)

shap_only = shap_df[feature_cols].loc[df.index].reset_index(drop=True)

df["abs_psr_change"] = df[TARGET_DELTA_COL].abs()

df["abs_rank_pct"] = df["abs_psr_change"].rank(pct=True)

df["shock_rank_pct"] = df["shock_score"].rank(pct=True)

df["priority_score"] = 0.7 df["abs_rank_pct"] + 0.3 df["shock_rank_pct"]

out_rows = []

for i in range(len(df)):

tmp = pd.DataFrame({

"feature": feature_cols,

"feature_delta": df.loc[i, feature_cols].values,

"contribution": shap_only.loc[i].values,

})

tmp["abs_contribution"] = tmp["contribution"].abs()

tmp = tmp.sort_values("abs_contribution", ascending=False).head(top_n_reasons)

reasons = [feature_to_reason(r["feature"], r["feature_delta"]) for _, r in tmp.iterrows()]

out_rows.append({

ID_COL: df.loc[i, ID_COL],

"after_date": df.loc[i, "after_date"],

TARGET_COL: df.loc[i, TARGET_COL] if TARGET_COL in df.columns else np.nan,

TARGET_LOG_COL: df.loc[i, TARGET_LOG_COL] if TARGET_LOG_COL in df.columns else np.nan,

TARGET_DELTA_COL: df.loc[i, TARGET_DELTA_COL],

TARGET_DELTA_LOG_COL: df.loc[i, TARGET_DELTA_LOG_COL],

"shock_score": df.loc[i, "shock_score"],

"abs_psr_change": df.loc[i, "abs_psr_change"],

"priority_score": df.loc[i, "priority_score"],

"reason_1": reasons[0] if len(reasons) > 0 else None,

"reason_2": reasons[1] if len(reasons) > 1 else None,

"reason_3": reasons[2] if len(reasons) > 2 else None,

})

out = pd.DataFrame(out_rows).sort_values(["priority_score", "abs_psr_change"], ascending=False).reset_index(drop=True)

return clip_inf_and_downcast(out)

# =========================================================

# EXCEL EXPORT

# =========================================================

def write_outputs_to_excel(

output_path: Path,

feature_store_latest: pd.DataFrame,

delta_store_eval: pd.DataFrame,

global_drivers: pd.DataFrame,

ops_worklist: pd.DataFrame,

facility_drivers: pd.DataFrame,

metrics: dict,

):

metrics_df = pd.DataFrame([{"metric": k, "value": v} for k, v in metrics.items()])

with pd.ExcelWriter(output_path, engine="openpyxl") as writer:

feature_store_latest.to_excel(writer, sheet_name="feature_store_latest", index=False)

delta_store_eval.to_excel(writer, sheet_name="delta_store_eval", index=False)

global_drivers.to_excel(writer, sheet_name="global_drivers", index=False)

ops_worklist.to_excel(writer, sheet_name="ops_worklist", index=False)

facility_drivers.to_excel(writer, sheet_name="facility_drivers", index=False)

metrics_df.to_excel(writer, sheet_name="model_metrics", index=False)

# =========================================================

# MAIN PIPELINE

# =========================================================

def run_pipeline(

train_start_date: str,

train_num_days: int,

eval_start_date: str,

base_path: Path = BASE_PATH,

output_xlsx: Path = OUTPUT_XLSX,

):

"""

Main runner.

Flow

----

1. Find available dates with all required files.

2. Split into train dates and eval dates.

3. Load raw daily trades + PSR.

4. Build facility-day datasets.

5. Build delta datasets.

6. Train on training deltas.

7. Score eval deltas.

8. Create ops worklist for latest eval day.

9. Export to Excel.

"""

# collect enough available dates to cover train and eval windows

available_dates = sorted([

d.name for d in base_path.iterdir()

if d.is_dir() and d.name.isdigit() and has_required_files(d.name, base_path)

])

if not available_dates:

raise ValueError("No valid dates found with required parquet files.")

train_dates, eval_dates = get_train_eval_dates(

available_dates=available_dates,

train_start_date=train_start_date,

train_num_days=train_num_days,

eval_start_date=eval_start_date,

)

all_needed_dates = sorted(set(train_dates) | set(eval_dates))

if len(train_dates) < 2:

raise ValueError("Need at least 2 available train dates.")

if len(eval_dates) < 2:

raise ValueError("Need at least 2 available eval dates including previous date before eval start.")

print("Train dates:", train_dates[0], "->", train_dates[-1], "count =", len(train_dates))

print("Eval dates :", eval_dates[0], "->", eval_dates[-1], "count =", len(eval_dates))

dfs, prs = load_daily_inputs(all_needed_dates, base_path)

# build facility-day store

facility_store_by_date = {}

for d in all_needed_dates:

print("Building facility-day dataset:", d)

facility_store_by_date[d] = build_facility_day_dataset(d, dfs[d], prs[d], base_path)

# build train delta store

train_delta_parts = []

for i in range(1, len(train_dates)):

d0, d1 = train_dates[i - 1], train_dates[i]

print("Building train delta:", d0, "->", d1)

train_delta_parts.append(make_delta_dataset(facility_store_by_date[d0], facility_store_by_date[d1]))

train_delta_store = clip_inf_and_downcast(pd.concat(train_delta_parts, ignore_index=True))

train_delta_store = add_shock_score(train_delta_store)

# build eval delta store

eval_delta_parts = []

for i in range(1, len(eval_dates)):

d0, d1 = eval_dates[i - 1], eval_dates[i]

print("Building eval delta:", d0, "->", d1)

eval_delta_parts.append(make_delta_dataset(facility_store_by_date[d0], facility_store_by_date[d1]))

eval_delta_store = clip_inf_and_downcast(pd.concat(eval_delta_parts, ignore_index=True))

eval_delta_store = add_shock_score(eval_delta_store)

# train model

train_model_df, X_train, y_train, feature_cols = build_model_matrix(train_delta_store)

model = train_xgb_model(X_train, y_train)

# evaluate on OOS

eval_model_df, X_eval, y_eval, = buildmodel_matrix(eval_delta_store)

pred_eval = model.predict(X_eval)

metrics = {

"train_rows": len(X_train),

"eval_rows": len(X_eval),

"n_features": len(feature_cols),

"delta_log_r2_oos": r2_score(y_eval, pred_eval),

"delta_log_mae_oos": mean_absolute_error(y_eval, pred_eval),

"train_start_date": train_start_date,

"train_num_days": train_num_days,

"eval_start_date": eval_start_date,

}

# reconstituted level check

if TARGET_COL in eval_model_df.columns and TARGET_DELTA_COL in eval_model_df.columns:

eval_check = eval_model_df[[ID_COL, "before_date", "after_date", TARGET_COL, TARGET_DELTA_COL]].copy()

eval_check["actual_level_today"] = eval_check[TARGET_COL] + eval_check[TARGET_DELTA_COL]

metrics["avg_abs_psr_change_oos"] = eval_check[TARGET_DELTA_COL].abs().mean()

# drivers on eval set

global_drivers, shap_eval = calculate_xgb_contributions(model, X_eval)

facility_drivers = get_facility_drivers(eval_model_df, X_eval, shap_eval, top_n=10)

latest_eval_after_date = eval_model_df["after_date"].max()

ops_worklist = build_ops_worklist(

scored_df=eval_model_df,

shap_df=shap_eval,

feature_cols=feature_cols,

latest_after_date=latest_eval_after_date,

top_n_reasons=3,

)

feature_store_latest = facility_store_by_date[ts_to_date_key(latest_eval_after_date)].copy()

write_outputs_to_excel(

output_path=output_xlsx,

feature_store_latest=feature_store_latest,

delta_store_eval=eval_delta_store,

global_drivers=global_drivers,

ops_worklist=ops_worklist,

facility_drivers=facility_drivers,

metrics=metrics,

)

return {

"train_dates": train_dates,

"eval_dates": eval_dates,

"facility_store_by_date": facility_store_by_date,

"train_delta_store": train_delta_store,

"eval_delta_store": eval_delta_store,

"model": model,

"global_drivers": global_drivers,

"facility_drivers": facility_drivers,

"ops_worklist": ops_worklist,

"metrics": metrics,

"output_xlsx": output_xlsx,

}

# =========================================================

# RUN

# =========================================================

# results = run_pipeline(

# train_start_date=TRAIN_START_DATE,

# train_num_days=TRAIN_NUM_DAYS,

# eval_start_date=EVAL_START_DATE,

# base_path=BASE_PATH,

# output_xlsx=OUTPUT_XLSX,

# )

# print(results["metrics"])

# print("Excel written to:", results["output_xlsx"])

Contact us

Whether you have a request, a query, or want to work with us, use the form below to get in touch with our team.