import os, re, glob, sys, argparse from datetime import datetime, timedelta import numpy as np import pandas as pd import PyUtil.PyRicX as RicX from PyUtil.PyLog import PyLog from PyUtil.PyDate import PyDate import PyUtil.PyRicX as RicX from openpyxl.styles import Alignment from openpyxl.styles import PatternFill from model.signal.interface.ShortAvailability import ShortAvailability from model.signal.engine.SignalMgr import SignalMgr PERISCOPE_EVENT_COLUMNS = [ "assetKey", "internal_ric", "ModelCntry", "code", "event_name", "review_cutoff_date", "announcement_date", "effective_date", "trade_date", "_trade_end_date", "_trade_end_anchor", "_trade_end_offset_days", "conviction", "side", "days_to_trade", "flow_usdm", "shares_to_trade", "trade_eligible", "_dtt_abs", ] def apply_periscope_overlay(rebalConfig, tradeDate, dframe): periscope_enabled = False tradeYmd = coerceToYYYYMMDD(tradeDate) if tradeYmd is None: PyLog.info(f"PERISCOPE: could not parse tradeDate= {tradeDate}. Skipping.") return dframe, periscope_enabled # create log dir and output dir for data run = create_periscope_dirs(tradeYmd) log = run["logger"] write_meta_sheet(run, tradeYmd, wb=None, status="initialized", note="created placeholder before reading vendor file") # Always create active_events sheet (even if empty). # Hard unwind reads previous trade day's active_events; # without this, disabled days produce invalid/missing sheets. write_active_events_sheet(run, None) log.info(f"PERISCOPE: start tradeDate={tradeYmd}") # read the relevant periscope file wb = read_periscope_file(run["input_dir"], tradeDate, log, skip_stale_file_check=True) if wb is None: msg = f"PERISCOPE: no input periscope file for {tradeYmd}; exiting gracefully." log.info(msg) write_meta_sheet(run, tradeYmd, wb=None, status="disabled_no_input_periscope_file", note=msg) close_periscope_files(run) return dframe, periscope_enabled run["df_change_raw"] = wb["df_change_raw"] write_periscope_detail_sheets(run, tradeYmd, wb) log.info(f"PERISCOPE: loaded periscope input file={wb['input_periscope_file']}") # just create standardized events with standardized column names etc df_events = build_periscope_events(wb["df_list_raw"]) write_dataframe_sheet( run, sheet_name="events_standardized", df=df_events, note_if_empty="no events", ) # map ric for china to our dataframe assetkeys taking into account that we do only northbound trading df_trade_map = build_trade_ric_map(dframe, log) # log all mapped and unmapped events df_events_mapped, df_events_unmapped = map_events_to_assetkey(df_events, df_trade_map, log) write_dataframe_sheet(run, sheet_name="mapped_events", df=df_events_mapped, note_if_empty="no mapped events", ) write_dataframe_sheet(run, sheet_name="unmapped_events", df=df_events_unmapped, note_if_empty="no unmapped events", ) params = get_periscope_params(rebalConfig) # Filter based on above periscope params like days_to_trade, conviction etc df_events_filtered = filter_events_for_trade_date(df_events_mapped, tradeDate, params, log) # write mapped events into excel write_dataframe_sheet(run, "events_mapped_filtered", df_events_filtered, "no filtered events") # for multile events per stock df_governing, df_aggregated = select_governing_events(df_events_filtered, log) # write to excel final events that we will proess and ignored events write_dataframe_sheet(run, "governing_events", df_governing, "no governing events") write_dataframe_sheet(run, "aggregated_events", df_aggregated, "no aggregated events") #create_trades = bool(getattr(rebalConfig, "createPeriscopeTrades", False)) dframe, eff_overlay = apply_periscope_alpha_and_bounds(rebalConfig, tradeYmd, dframe, df_governing, log) dframe, eff_unwind, df_dropped = apply_periscope_hard_unwind(rebalConfig, tradeYmd, run, dframe, df_governing, log) write_dataframe_sheet(run, "dropped_events", df_dropped, "no dropped events") write_active_events_sheet(run, df_governing) df_optimizer_overrides = write_optimizer_overrides_sheet(run, eff_overlay, eff_unwind) # this is additional step to create manual trades based on long/short logic # Currently the process to create periscope trades is manual, we have not integrated # periscope into our optimizer with base strategy or standalone separate optimizer # So this step is needed. Once we have one of the above this step can be eliminated. res = build_manual_periscope_portfolio( trade_date_yyyymmdd=tradeYmd, periscope_trade_gross_usd=12_500_000, max_short_utilization=0.25, df_optimizer_overrides =df_optimizer_overrides ) for sheet_name, df_ in res.items(): write_dataframe_sheet(run, sheet_name, df_, "no data") preferred = [ "final_trades", "borrow", "restrictions", "optimizer_overrides", "active_events", "dropped_events", "list_raw", "change_raw", "governing_events", "aggregated_events", "events_mapped_filtered", "mapped_events", "unmapped_events", "events_standardized", "meta", "process_logic", ] reorder_output_sheets(run, preferred_order=preferred, log=run["logger"]) # post processing close_periscope_files(run) periscope_enabled = True return dframe, periscope_enabled def build_trade_ric_map(dframe, log): mfm = get_model_universe_frame() df_trade = dframe[["assetKey"]].copy() df_trade = df_trade.merge(mfm[["assetKey", "internal_ric", "ModelCntry"]], on="assetKey", how="left", sort=False ) """ df_trade["internal_ric"] = df_trade["internal_ric"].astype(str).str.strip().str.upper() df_trade["match_key"] = df_trade["internal_ric"].apply(_match_key_from_ric) missing = df_trade["internal_ric"].isna().sum() if missing: log.info(f"PERISCOPE: warning: {missing} rows in dframe have no internal_ric in model_universe_frame") """ # count missing BEFORE astype(str), otherwise NaN becomes "nan" missing = df_trade["internal_ric"].isna().sum() if missing: log.info(f"PERISCOPE: warning: {missing} rows in dframe have no internal_ric in model_universe_frame") df_trade["internal_ric"] = df_trade["internal_ric"].astype(str).str.strip().str.upper() df_trade["match_key"] = df_trade["internal_ric"].apply(_match_key_from_ric) dupe = df_trade.groupby("match_key")["assetKey"].nunique() dupe_keys = set(dupe[dupe > 1].index.tolist()) if dupe_keys: log.info(f"PERISCOPE: warning: {len(dupe_keys)} match_key values map to multiple assetKey in today's dframe; keeping first") df_trade = df_trade.dropna(subset=["match_key"]) df_trade = df_trade.drop_duplicates(subset=["match_key"], keep="first") return df_trade[["match_key", "assetKey", "internal_ric", "ModelCntry"]] def map_events_to_assetkey(df_events, df_trade_map, log): if df_events is None or df_events.empty: return pd.DataFrame(), pd.DataFrame() ev = df_events.copy() ev["code"] = ev["code"].astype(str).str.strip().str.upper() ev["match_key"] = ev["code"].apply(_match_key_from_ric) mapped = ev.merge(df_trade_map[["match_key", "assetKey", "internal_ric", "ModelCntry"]], on="match_key", how="left", sort=False) mapped_ok = mapped[mapped["assetKey"].notna()].copy() unmapped = mapped[mapped["assetKey"].isna()].copy() cols = list(mapped_ok.columns) front = [] for c in ["code", "internal_ric", "assetKey"]: if c in cols: front.append(c) cols.remove(c) mapped_ok = mapped_ok[front + cols] log.info(f"PERISCOPE: events total={len(ev)}, mapped_to_dframe={len(mapped_ok)}, unmapped={len(unmapped)}") return mapped_ok, unmapped def filter_events_for_trade_date(df_events_mapped, tradeDate, params, log): if df_events_mapped is None or df_events_mapped.empty: return pd.DataFrame() df = df_events_mapped.copy() excluded = params.get("excluded_countries", set()) if excluded: df = df[~df["ModelCntry"].astype(str).str.strip().str.upper().isin(excluded)].copy() trade_date = pd.to_datetime(tradeDate).normalize() df["trade_date"] = trade_date df["_effective_date"] = pd.to_datetime(df["effective_date"], errors="coerce") df["_announcement_date"] = pd.to_datetime(df["announcement_date"], errors="coerce") df["_cutoff_date"] = pd.to_datetime(df["review_cutoff_date"], errors="coerce") rules = params.get("event_rules") if not rules: raise ValueError("PERISCOPE: event_rules missing in params") picked_rules = [] for ev in df["event_name"].fillna("").astype(str): chosen_rule = None for r in rules: pat = r.get("pattern") if pat and re.search(pat, ev, flags=re.IGNORECASE): chosen_rule = r break if chosen_rule is None: chosen_rule = next( (r for r in rules if r.get("pattern") is None), None ) if chosen_rule is None: raise ValueError("PERISCOPE: DEFAULT rule missing in event_rules") picked_rules.append(chosen_rule) df["_event_rule"] = [r["key"] for r in picked_rules] df["_trade_end_anchor"] = [r["trade_end_anchor"] for r in picked_rules] df["_trade_end_offset_days"] = [ int(r.get("trade_end_offset_days", 0) or 0) for r in picked_rules ] df["_trade_start_date"] = trade_date df["_trade_end_anchor_date"] = pd.NaT m = df["_trade_end_anchor"] == "effective" df.loc[m, "_trade_end_anchor_date"] = df.loc[m, "_effective_date"] m = df["_trade_end_anchor"] == "announcement" df.loc[m, "_trade_end_anchor_date"] = df.loc[m, "_announcement_date"] m = df["_trade_end_anchor"] == "cutoff" df.loc[m, "_trade_end_anchor_date"] = df.loc[m, "_cutoff_date"] bad = ~df["_trade_end_anchor"].isin(["effective", "announcement", "cutoff"]) if bad.any(): raise ValueError( f"PERISCOPE: unknown trade_end_anchor values: " f"{sorted(df.loc[bad, '_trade_end_anchor'].unique())}" ) df["_trade_end_date"] = ( df["_trade_end_anchor_date"] + pd.to_timedelta(df["_trade_end_offset_days"], unit="D") ) df = df[df["_trade_end_date"].notna()].copy() df = df[trade_date <= df["_trade_end_date"]].copy() allowed = params["allowed_convictions"] df["_conviction_norm"] = df["conviction"].astype(str).str.strip() df = df[df["_conviction_norm"].isin(allowed)].copy() dtt_abs = pd.to_numeric(df["days_to_trade"], errors="coerce").abs() df["_dtt_abs"] = dtt_abs interest_window_days = params.get("interest_window_days", 0) interest_end = trade_date + pd.Timedelta(days=interest_window_days) df = df[ (df["_trade_end_date"].notna()) & (df["_trade_end_date"] >= trade_date) & (df["_trade_end_date"] <= interest_end) ].copy() df["trade_eligible"] = ( dtt_abs.notna() & (dtt_abs >= params["min_dtt"]) & (df["_trade_end_date"] >= trade_date) & (df["_trade_end_date"] <= interest_end) ) log.info( f"PERISCOPE: filtered events count={len(df)}, " f"min_dtt={params['min_dtt']}, " f"allowed={sorted(list(allowed))}, " f"trade_eligible={int(df['trade_eligible'].sum())}" ) return df def select_governing_events(df_filtered, log): """ Option A (Netting via FLOW): - For each assetKey, net signed flow_usdm across all ACTIVE rows (df_filtered). (flow_usdm is assumed present + already signed.) - If net flow == 0 => no governing trade for that assetKey (still reported in aggregated sheet). - If net flow != 0 => choose ONE representative row to carry event metadata, then overwrite: flow_usdm = net_flow_usdm side = sign(net_flow) (side is +/- 1) shares_to_trade = net_shares (kept for info only, NOT used for net direction) Special case (requested): - If all INCLUDED legs are in the same direction (all flow_usdm have same sign), choose the representative row with the LARGEST abs(days_to_trade)). Announced handling (per our discussion): - EXCLUDE rows with conviction == "Announced" from NETTING *only if* _trade_end_anchor != "announcement". (If _trade_end_anchor == "announcement", we KEEP "Announced" rows in netting.) """ if df_filtered is None or df_filtered.empty: return pd.DataFrame(), pd.DataFrame() df = df_filtered.copy() # --- Required columns (we rely on standardize_event_columns already) --- required = ["assetKey", "conviction", "effective_date", "days_to_trade", "flow_usdm", "side"] missing = [c for c in required if c not in df.columns] if missing: raise ValueError(f"PERISCOPE: df_filtered missing columns: {missing}") # Helpers used for selection + reporting df["_conv_rank"] = df["conviction"].apply(conviction_rank) df["_effective_date"] = pd.to_datetime(df["effective_date"], errors="coerce") df["_dtt_abs"] = pd.to_numeric(df["days_to_trade"], errors="coerce").abs() # FLOW is the netting driver (SIGNED) df["_flow_signed"] = pd.to_numeric(df["flow_usdm"], errors="coerce") df["_flow_abs"] = df["_flow_signed"].abs() # Shares are informational only (optional) if "shares_to_trade" in df.columns: df["_shares_signed"] = pd.to_numeric(df["shares_to_trade"], errors="coerce") else: df["_shares_signed"] = pd.NA # Count legs per assetKey (how many events per name) df["event_count"] = df.groupby("assetKey")["assetKey"].transform("size") governing_rows = [] aggregated_rows = [] # Work group-by without resorting original ordering elsewhere in pipeline for asset_key, g in df.groupby("assetKey", sort=False): g = g.copy() # ------------------------------------------------------------ # Decide which legs participate in NETTING # ------------------------------------------------------------ conv_norm = g["conviction"].astype(str).str.strip() is_announced = conv_norm.eq("Announced") # If _trade_end_anchor exists, only exclude Announced when anchor != announcement if "_trade_end_anchor" in g.columns: anchor_norm = g["_trade_end_anchor"].astype(str).str.strip().str.lower() exclude_announced_from_netting = is_announced & (anchor_norm != "announcement") else: exclude_announced_from_netting = pd.Series([False] * len(g), index=g.index) g_netting = g[~exclude_announced_from_netting].copy() # Net signed FLOW (only from included legs) if g_netting.empty: net_flow = 0.0 net_shares = 0.0 else: net_flow = g_netting["_flow_signed"].sum(skipna=True) # informational only net_shares = ( g_netting["_shares_signed"].sum(skipna=True) if "_shares_signed" in g_netting.columns else 0.0 ) # Determine net direction from FLOW if pd.isna(net_flow) or float(net_flow) == 0.0: net_side = 0 else: net_side = 1 if net_flow > 0 else -1 # ------------------------------------------------------------ # Build one-row aggregated summary (unique per assetKey) # (summary uses ALL legs, netting uses g_netting) # ------------------------------------------------------------ g_sorted_for_summary = g.sort_values( by=["_conv_rank", "_effective_date", "_dtt_abs"], ascending=[False, True, False], na_position="last", kind="mergesort", ) rep = g_sorted_for_summary.iloc[0].copy() rep["net_flow_usdm"] = net_flow rep["net_shares_to_trade"] = net_shares # info only rep["net_side"] = net_side rep["netting_leg_count"] = int(len(g)) rep["netting_included_leg_count"] = int(len(g_netting)) rep["netting_excluded_announced_count"] = int(exclude_announced_from_netting.sum()) # Compact leg summary (all legs) using FLOW if "event_name" in g.columns: legs = [] for _, r in g.iterrows(): en = str(r.get("event_name", "") or "") fl = r.get("_flow_signed") if pd.isna(fl): legs.append(f"{en}:NA") else: legs.append(f"{en}:{float(fl):.3f}") rep["netting_legs_summary"] = " | ".join(legs) else: rep["netting_legs_summary"] = "" if net_side == 0: rep["netting_reason"] = "NET_ZERO" aggregated_rows.append(rep) continue # ------------------------------------------------------------ # Determine same-direction using INCLUDED legs only (FLOW signs) # ------------------------------------------------------------ signs = g_netting["_flow_signed"].dropna() signs = signs[signs != 0] same_direction = False if len(signs) > 0: same_direction = (np.all(signs > 0) or np.all(signs < 0)) rep["netting_same_direction"] = bool(same_direction) # ------------------------------------------------------------ # Pick governing representative row used for metadata # - only consider INCLUDED legs that match net FLOW direction # ------------------------------------------------------------ g_net = g_netting[ g_netting["_flow_signed"].notna() & (g_netting["_flow_signed"] * net_side > 0) ].copy() if g_net.empty: # Rare: net != 0 but can't find matching-direction included legs (data issues) chosen = rep.copy() rep["netting_reason"] = "NET_NONZERO_FALLBACK_REP" else: if same_direction: # Requested special case: choose largest abs(days_to_trade) g_net = g_net.sort_values( by=["_dtt_abs", "_conv_rank", "_effective_date"], ascending=[False, False, True], na_position="last", kind="mergesort", ) chosen = g_net.iloc[0].copy() rep["netting_reason"] = "SAME_DIR_PICK_MAX_DTT" else: # Mixed directions: pick biggest abs(FLOW) on net side, # tie-break by conviction then effective then dtt_abs g_net = g_net.sort_values( by=["_flow_abs", "_conv_rank", "_effective_date", "_dtt_abs"], ascending=[False, False, True, False], na_position="last", kind="mergesort", ) chosen = g_net.iloc[0].copy() rep["netting_reason"] = "MIXED_DIR_PICK_MAX_ABS_FLOW" # ------------------------------------------------------------ # Build governing row = chosen metadata + overwrite trade instruction with NET # ------------------------------------------------------------ gov = chosen.copy() gov["flow_usdm"] = net_flow gov["side"] = net_side # shares_to_trade: keep consistent with net_shares for info ONLY if "shares_to_trade" in gov.index: gov["shares_to_trade"] = net_shares # Counters gov["event_count"] = int(len(g)) gov["ignored_event_count"] = int(max(len(g) - 1, 0)) # Netting context gov["netting_leg_count"] = int(len(g)) gov["netting_included_leg_count"] = int(len(g_netting)) gov["netting_excluded_announced_count"] = int(exclude_announced_from_netting.sum()) gov["netting_reason"] = rep["netting_reason"] gov["netting_legs_summary"] = rep.get("netting_legs_summary", "") governing_rows.append(gov) aggregated_rows.append(rep) df_governing = pd.DataFrame(governing_rows) df_aggregated = pd.DataFrame(aggregated_rows) log.info( f"PERISCOPE: netting governing selection (FLOW): " f"assets_in={df['assetKey'].nunique()} " f"governing_assets={df_governing['assetKey'].nunique() if not df_governing.empty else 0} " f"aggregated_assets={df_aggregated['assetKey'].nunique() if not df_aggregated.empty else 0}" ) return df_governing, df_aggregated def apply_periscope_alpha_and_bounds(rebalConfig, tradeYmd, dframe, df_governing_events, log): if df_governing_events is None or df_governing_events.empty: return dframe, pd.DataFrame() for c in ["assetKey", "alpha", "longBoundHard", "shortBoundHard", "preOptWeights"]: if c not in dframe.columns: raise ValueError(f"dframe missing {c}") for c in ["assetKey", "side", "conviction"]: if c not in df_governing_events.columns: raise ValueError(f"df_governing_events missing {c}") alpha_high = getattr(rebalConfig, "periscopeAlphaHigh", 0.15) # 1.25% GMV with leverage=4 => 5% NAV weight cap max_abs_w = float(getattr(rebalConfig, "periscopeMaxAbsWeight", 0.05)) # --- IMPORTANT CHANGE: only apply overlay to trade_eligible names --- ev = df_governing_events.copy() event_map = ev.set_index("assetKey")["event_name"] dframe["periscopeEvent"] = dframe["assetKey"].map(event_map).fillna("UNKNOWN_EVENT") # ------------------------------------------------------------------ # Mark ACTIVE periscope names (for portfolio-level gross constraint) # Includes ALL lifecycle-valid governing names # ------------------------------------------------------------------ dframe["isPeriscopeActive"] = False active_keys = ev["assetKey"].astype(str).unique() dframe.loc[ dframe["assetKey"].astype(str).isin(active_keys), "isPeriscopeActive" ] = True if "trade_eligible" in ev.columns: ev["trade_eligible"] = ev["trade_eligible"].fillna(False).astype(bool) ev_trade = ev[ev["trade_eligible"]].copy() else: # backward-compatible default ev_trade = ev.copy() ev_trade["trade_eligible"] = True if ev_trade.empty: # nothing eligible to trade today => no overlay today return dframe, pd.DataFrame() ev_trade["conviction"] = ev_trade["conviction"].astype(str).str.strip().str.lower() ev_trade["side_num"] = pd.to_numeric(ev_trade["side"], errors="coerce") ev_trade = ev_trade[ev_trade["side_num"].notna()].copy() if "flow_usdm" not in ev_trade.columns: raise ValueError("PERISCOPE: flow_usdm missing in governing events (needed for flow-based alpha)") ##### RANK SIZING ###### flow = pd.to_numeric(ev_trade["flow_usdm"], errors="coerce") flow_abs = flow.abs().fillna(0.0) ################# END RANK SIZING ################# # Percentile rank in [0,1]; if only 1 row -> 1.0 if len(flow_abs) == 1: pct = pd.Series([1.0], index=ev_trade.index) else: pct = flow_abs.rank(method="average", pct=True) # alpha magnitude in [0, alpha_high] ev_trade["alpha_mag"] = (pct * float(alpha_high)).clip(lower=0.0, upper=float(alpha_high)) # signed alpha ev_trade["alpha_peri"] = ev_trade["alpha_mag"] * ev_trade["side_num"] ################# END RANK SIZING ################# """ ############### FLOW PROPORTIONAL SIZING ##################### flow = pd.to_numeric(ev_trade["flow_usdm"], errors="coerce").fillna(0.0) flow_abs = flow.abs() den = flow_abs.sum() if den <= 0: ev_trade["alpha_mag"] = 0.0 else: # flow-proportional weights, then cap per-name ev_trade["alpha_mag"] = (flow_abs / den) * float(alpha_high) ev_trade["alpha_mag"] = ev_trade["alpha_mag"].clip(lower=0.0, upper=float(alpha_high)) ev_trade["alpha_peri"] = ev_trade["alpha_mag"] * ev_trade["side_num"] ############### END FLOW PROPORTIONAL SIZING ##################### """ alpha_by_asset = dict(zip(ev_trade["assetKey"].astype(str), ev_trade["alpha_peri"])) side_by_asset = dict(zip(ev_trade["assetKey"].astype(str), ev_trade["side_num"])) asset = dframe["assetKey"].astype(str) mask = asset.isin(alpha_by_asset.keys()) # Mark periscope names touched today (trade-eligible periscope only) # Very very important mate # Mark periscope names touched today (trade-eligible only) dframe["isPeriscopeTrade"] = False dframe.loc[mask, "isPeriscopeTrade"] = True if not mask.any(): return dframe, pd.DataFrame() # snapshot pre alpha_pre = dframe.loc[mask, "alpha"].copy() lb_pre = dframe.loc[mask, "longBoundHard"].copy() sb_pre = dframe.loc[mask, "shortBoundHard"].copy() prew = pd.to_numeric(dframe.loc[mask, "preOptWeights"], errors="coerce").fillna(0.0) # apply alpha replacement dframe.loc[mask, "alpha"] = asset.loc[mask].map(alpha_by_asset).to_numpy() # force sign only side_today = asset.loc[mask].map(side_by_asset) mask_long = mask & (side_today > 0) mask_short = mask & (side_today < 0) # LONG: block shorts + cap long bound if mask_long.any(): dframe.loc[mask_long, "shortBoundHard"] = 0.0 dframe.loc[mask_long, "longBoundHard"] = ( pd.to_numeric(dframe.loc[mask_long, "longBoundHard"], errors="coerce") .fillna(0.0) .clip(upper=max_abs_w) ) # SHORT: block longs + cap short bound (shortBoundHard is positive in the system) if mask_short.any(): dframe.loc[mask_short, "longBoundHard"] = 0.0 dframe.loc[mask_short, "shortBoundHard"] = ( pd.to_numeric(dframe.loc[mask_short, "shortBoundHard"], errors="coerce") .fillna(0.0) .clip(upper=max_abs_w) ) # snapshot post alpha_post = dframe.loc[mask, "alpha"].copy() lb_post = dframe.loc[mask, "longBoundHard"].copy() sb_post = dframe.loc[mask, "shortBoundHard"].copy() # build effects sheet rows (include full event context) keep = [c for c in PERISCOPE_EVENT_COLUMNS if c in ev_trade.columns] df_event_details = ev_trade[keep].copy() df_event_details["assetKey"] = df_event_details["assetKey"].astype(str) df_event_details = df_event_details.drop_duplicates(subset=["assetKey"], keep="first") eff = pd.DataFrame({ "assetKey": asset.loc[mask].to_numpy(), "preOptWeights": prew.to_numpy(), "alpha_pre": alpha_pre.to_numpy(), "alpha_post": alpha_post.to_numpy(), "longBoundHard_pre": lb_pre.to_numpy(), "longBoundHard_post": lb_post.to_numpy(), "shortBoundHard_pre": sb_pre.to_numpy(), "shortBoundHard_post": sb_post.to_numpy(), }) # prevent _x/_y explosion: drop overlapping columns from ev_ctx overlap = (set(eff.columns) & set(df_event_details.columns)) - {"assetKey"} if overlap: df_event_details = df_event_details.drop(columns=sorted(list(overlap))) eff = eff.merge(df_event_details, on="assetKey", how="left") eff["reason"] = eff["side"].apply(lambda x: "GOVERNING_LONG" if float(x) > 0 else "GOVERNING_SHORT") log.info( f"PERISCOPE: overlay applied tradeYmd={tradeYmd} n_touched={int(mask.sum())} " f"maxAbsWeight={max_abs_w}" ) return dframe, eff def apply_periscope_hard_unwind(rebalConfig, tradeYmd, run, dframe, df_governing_events, log): enabled = bool(getattr(rebalConfig, "periscopeHardUnwindEnabled", True)) if not enabled: return dframe, pd.DataFrame(), pd.DataFrame() # default dropped sheet df_dropped = pd.DataFrame(columns=["assetKey", "tradeYmd"]) for c in ["assetKey", "preOptWeights", "longBoundHard", "shortBoundHard", "alpha"]: if c not in dframe.columns: raise ValueError(f"dframe missing {c}") # today's active set (governing events are already lifecycle-filtered upstream) today_active = set() if df_governing_events is not None and not df_governing_events.empty: today_active = set(df_governing_events["assetKey"].astype(str).tolist()) # previous trading-day details file prev_path = get_previous_details_path(run["output_data_file"], tradeYmd) if not prev_path or (not os.path.isfile(prev_path)): log.info(f"PERISCOPE: hard unwind skipped (missing prev weekday details): {prev_path}") return dframe, pd.DataFrame(), df_dropped prev_active_df = read_active_events(prev_path) if prev_active_df.empty: log.info("PERISCOPE: hard unwind skipped (prev weekday active_events empty)") return dframe, pd.DataFrame(), df_dropped prev_active_df = prev_active_df.copy() prev_active_df["assetKey"] = prev_active_df["assetKey"].astype(str) prev_active = set(prev_active_df["assetKey"].tolist()) dropped = prev_active - today_active if not dropped: log.info(f"PERISCOPE: hard unwind tradeYmd={tradeYmd} dropped=0") return dframe, pd.DataFrame(), df_dropped # ---- Build dropped_events sheet with full context (from prev_active_df) ---- df_dropped = prev_active_df[prev_active_df["assetKey"].isin(dropped)].copy() df_dropped["tradeYmd"] = tradeYmd # Optional: add vendor Change column from today's change_raw, if available df_change_raw = run.get("df_change_raw", None) if df_change_raw is not None and not df_change_raw.empty and "code" in df_dropped.columns: ch = df_change_raw.copy() # drop vendor's blank separator rows (between "new" and "drop") ch = ch.dropna(how="all") # normalize vendor column names (e.g. "Code", "Change", trailing spaces) ch.columns = [str(c).strip().lower() for c in ch.columns] # require vendor columns after normalization if "code" in ch.columns and "change" in ch.columns: # normalize join keys df_dropped["code"] = df_dropped["code"].astype(str).str.strip() ch["code"] = ch["code"].astype(str).str.strip() # keep one row per code (last wins), then merge ch2 = ( ch[["code", "change"]] .dropna(subset=["code"]) .drop_duplicates(subset=["code"], keep="last") .rename(columns={"change": "Change"}) ) df_dropped = df_dropped.merge(ch2, on="code", how="left") # Move Change right after event_name if "Change" in df_dropped.columns and "event_name" in df_dropped.columns: cols = list(df_dropped.columns) cols.remove("Change") cols.insert(cols.index("event_name") + 1, "Change") df_dropped = df_dropped[cols] # ---- Now apply hard unwind to dframe (only for dropped names that exist in dframe) ---- asset = dframe["assetKey"].astype(str) prew = pd.to_numeric(dframe["preOptWeights"], errors="coerce").fillna(0.0) mask_drop = asset.isin(dropped) if not mask_drop.any(): log.info(f"PERISCOPE: hard unwind tradeYmd={tradeYmd} dropped_in_dframe=0") return dframe, pd.DataFrame(), df_dropped if "event_name" not in df_dropped.columns: raise ValueError("PERISCOPE: df_dropped missing event_name for hard unwind") drop_event_map = df_dropped.drop_duplicates(subset=["assetKey"], keep="first") \ .set_index("assetKey")["event_name"] dframe.loc[mask_drop, "periscopeEvent"] = asset.loc[mask_drop].map(drop_event_map) # snapshot pre alpha_pre = dframe.loc[mask_drop, "alpha"].copy() lb_pre = dframe.loc[mask_drop, "longBoundHard"].copy() sb_pre = dframe.loc[mask_drop, "shortBoundHard"].copy() prew_drop = prew.loc[mask_drop].copy() # force flatten-to-zero mask_long = mask_drop & (prew > 0) mask_short = mask_drop & (prew < 0) if mask_long.any(): dframe.loc[mask_long, "longBoundHard"] = 0.0 if mask_short.any(): dframe.loc[mask_short, "shortBoundHard"] = 0.0 # snapshot post alpha_post = dframe.loc[mask_drop, "alpha"].copy() lb_post = dframe.loc[mask_drop, "longBoundHard"].copy() sb_post = dframe.loc[mask_drop, "shortBoundHard"].copy() eff = pd.DataFrame({ "assetKey": asset.loc[mask_drop].to_numpy(), "preOptWeights": prew_drop.to_numpy(), "alpha_pre": alpha_pre.to_numpy(), "alpha_post": alpha_post.to_numpy(), "longBoundHard_pre": lb_pre.to_numpy(), "longBoundHard_post": lb_post.to_numpy(), "shortBoundHard_pre": sb_pre.to_numpy(), "shortBoundHard_post": sb_post.to_numpy(), }) # attach previous active event context (full row) + optional Change column df_prev_active_events = df_dropped.drop(columns=["tradeYmd"], errors="ignore").copy() df_prev_active_events["assetKey"] = df_prev_active_events["assetKey"].astype(str) # enforce global periscope schema ordering keep = [c for c in PERISCOPE_EVENT_COLUMNS if c in df_prev_active_events.columns] df_prev_active_events = df_prev_active_events[keep] # prevent _x/_y explosion: drop overlapping columns from right side overlap = (set(eff.columns) & set(df_prev_active_events.columns)) - {"assetKey"} if overlap: df_prev_active_events = df_prev_active_events.drop(columns=sorted(list(overlap))) eff = eff.merge(df_prev_active_events, on="assetKey", how="left") def _reason(pw): if pw > 0: return "DROPPED_UNWIND_LONG_TO_0" if pw < 0: return "DROPPED_UNWIND_SHORT_TO_0" return "DROPPED_ALREADY_FLAT" eff["reason"] = eff["preOptWeights"].apply(_reason) log.info( f"PERISCOPE: hard unwind tradeYmd={tradeYmd} " f"dropped_total={len(dropped)} dropped_in_dframe={int(mask_drop.sum())} " f"forced_long_to0={int(mask_long.sum())} forced_short_to0={int(mask_short.sum())}" ) return dframe, eff, df_dropped def read_periscope_file(inputDir, tradeDate, log, skip_stale_file_check=False): if (not inputDir) or (not os.path.isdir(inputDir)): msg = f"PERISCOPE: input dir missing or not a directory: {inputDir}, Skipping." log.info(msg) return None tradeYmd = coerceToYYYYMMDD(tradeDate) if tradeYmd is None: msg = f"PERISCOPE: could not parse tradeDate={tradeDate}. Skipping." log.info(msg) return None periscope_file = find_latest_periscope_file(inputDir, tradeYmd, log) if periscope_file is None: log.info( f"PERISCOPE: no eligible Rebalance_Summary_YYYYMMDD.xlsx with date <= {tradeYmd} " f"found in {inputDir}. Skipping overlay." ) return None fileDateYmd = periscope_file["file_date_yyyymmdd"] filePath = periscope_file["path"] log.info(f"PERISCOPE: loaded periscope input file {filePath}") ageDays = (pd.to_datetime(tradeYmd) - pd.to_datetime(fileDateYmd)).days if not skip_stale_file_check and ageDays > 7: msg = f"PERISCOPE: latest file is stale ({ageDays} days old): {periscope_file['path']}. Disabling." log.info(msg) return None elif skip_stale_file_check and ageDays > 7: log.info(f"PERISCOPE: stale file allowed ({ageDays} days old): {periscope_file['path']}") most_recent_periscope_file = periscope_file["path"] file_excel = pd.ExcelFile(most_recent_periscope_file) sheetNames = file_excel.sheet_names # list sheet logic listSheet = find_sheet(sheetNames, "list") if listSheet is None: if len(sheetNames) == 1: listSheet = sheetNames[0] else: sheet1 = find_sheet(sheetNames, "Sheet1") if sheet1 is not None: listSheet = sheet1 # change sheet logic (support both change / changes) changeSheet = find_sheet(sheetNames, "change") if changeSheet is None: changeSheet = find_sheet(sheetNames, "changes") dfList = read_sheet(most_recent_periscope_file, listSheet) if listSheet else None dfChange = read_sheet(most_recent_periscope_file, changeSheet) if changeSheet else None log.info(f"PERISCOPE: loaded periscope input file {most_recent_periscope_file} (list={dfList is not None}, change={dfChange is not None})") return { "input_periscope_file": most_recent_periscope_file, "sheet_names": sheetNames, "df_list_raw": dfList, "df_change_raw": dfChange, "file_date_yyyymmdd": fileDateYmd, "age_days": ageDays, } def find_latest_periscope_file(inputDir, tradeYmd, log): """ Return the latest Rebalance_Summary_YYYYMMDD.xlsx with file_date <= tradeYmd. If none exists, return None (NO fallback to future/latest file). """ # normalize tradeYmd to 'YYYYMMDD' (handles int, '20210104', '2021-01-04', datetime.date, etc.) trade_str = re.sub(r"\D", "", str(tradeYmd)) if len(trade_str) != 8: log.info(f"PERISCOPE: bad tradeYmd={tradeYmd} (normalized={trade_str}); skipping overlay.") return None pat = re.compile(r"^Rebalance_Summary_(\d{8})\.xlsx$", re.IGNORECASE) files = [] for fn in os.listdir(inputDir): m = pat.match(fn) if not m: continue fdate = m.group(1) # 'YYYYMMDD' files.append((fdate, os.path.join(inputDir, fn))) if not files: return None # sort so "previous to trade date" is well-defined files.sort(key=lambda x: x[0]) eligible = [x for x in files if x[0] <= trade_str] if not eligible: log.info(f"PERISCOPE: no input periscope file for {trade_str}; exiting gracefully.") return None chosen = eligible[-1] return {"file_date_yyyymmdd": chosen[0], "path": chosen[1]} def find_sheet(sheetNames, targetLower): tgt = str(targetLower).strip().lower() for sn in sheetNames: if str(sn).strip().lower() == tgt: return sn return None def read_sheet(path, sheetName): raw = pd.read_excel(path, sheet_name=sheetName, header=None) hdr = find_header_row(raw, requiredCols=["Code"]) # Support old style sheet from 2021 if hdr == 0: hdr = find_header_row(raw, requiredCols=["RIC"]) return pd.read_excel(path, sheet_name=sheetName, header=hdr) def find_header_row(rawDf, requiredCols): req = [str(c).strip().lower() for c in requiredCols] maxScan = min(80, len(rawDf)) for r in range(maxScan): rowVals = [str(x).strip().lower() for x in rawDf.iloc[r].tolist()] ok = True for c in req: if not any(rv == c for rv in rowVals): ok = False break if ok: return r return 0 def create_periscope_dirs(tradeYmd): baseDir = os.environ.get("PERISCOPE_DATA_DIR", "").strip() logDir = os.environ.get("PERISCOPE_LOG_DIR", "").strip() if not baseDir: baseDir = "/data/periscope" if not logDir: logDir = "/data/log/periscope" inputDir = os.path.join(baseDir, "input") outDir = os.path.join(baseDir, "output") os.makedirs(inputDir, exist_ok=True) os.makedirs(outDir, exist_ok=True) os.makedirs(logDir, exist_ok=True) output_data_file = os.path.join(outDir, "periscope_details_trade_date_%s.xlsx" % tradeYmd) log_file = os.path.join(logDir, "periscope_trade_date_%s.log" % tradeYmd) logger = PeriscopeLogger(log_file) writer = pd.ExcelWriter(output_data_file, engine="openpyxl", datetime_format="yyyy-mm-dd", date_format="yyyy-mm-dd") return { "input_dir": inputDir, "output_data_file": output_data_file, "log_file": log_file, "logger": logger, "writer": writer, } class PeriscopeLogger: def __init__(self, path): self.path = path self.fh = open(path, "w", encoding="utf-8") def info(self, msg): PyLog.info(msg) self.fh.write(msg + "\n") self.fh.flush() def error(self, msg): PyLog.info(msg) self.fh.write("ERROR: " + msg + "\n") self.fh.flush() def close(self): try: self.fh.close() except Exception: pass def close_periscope_files(run): try: if run.get("writer") is not None: run["writer"].close() except Exception: pass try: if run.get("logger") is not None: run["logger"].close() except Exception: pass def coerceToYYYYMMDD(d): if d is None: return None if isinstance(d, str): s = d.strip() if re.fullmatch(r"\d{8}", s): return s try: return pd.to_datetime(s).strftime("%Y%m%d") except Exception: return None try: return pd.to_datetime(d).strftime("%Y%m%d") except Exception: return None def build_periscope_events(df_list_raw): if df_list_raw is None or df_list_raw.empty: return pd.DataFrame() df = df_list_raw.copy() df = standardize_event_columns(df) df = compute_side(df) df = coerce_event_types(df) df["code"] = df["code"].astype(str).str.strip().str.upper() df = df[df["code"].notna() & (df["code"] != "")] return df def standardize_event_columns(df): if df is None or df.empty: raise ValueError("PERISCOPE: input vendor file is empty") out = df.copy() out.columns = [str(c).strip() for c in out.columns] rename = { "Code": "code", "RIC": "code", "Ric": "code", "SEDOL": "sedol", "Name": "name", "Conviction": "conviction", "Review Cut-off date": "review_cutoff_date", "Review Cut-off Date": "review_cutoff_date", "Announcement Date": "announcement_date", "Effective Date": "effective_date", "Event Name": "event_name", "Side": "side", # flows "Flow US$m": "flow_usdm", # already in USD mm "Flow in USD": "flow_usd", # absolute USD, we’ll convert to mm below "Flow USD": "flow_usd", "Shares to Trade": "shares_to_trade", "Days to Trade": "days_to_trade", "DaysToTrade": "days_to_trade", "SharesToTrade": "shares_to_trade", } out = out.rename(columns={k: v for k, v in rename.items() if k in out.columns}) # If vendor provides Flow in USD, convert to US$mm to preserve your existing "flow_usdm" usage. if "flow_usdm" not in out.columns and "flow_usd" in out.columns: flow = pd.to_numeric(out["flow_usd"].astype(str).str.replace(",", ""), errors="coerce") out["flow_usdm"] = flow / 1e6 # Make numeric fields actually numeric (handles commas like "12,021,884") for c in ["side", "flow_usdm", "shares_to_trade", "days_to_trade"]: if c in out.columns: out[c] = pd.to_numeric(out[c].astype(str).str.replace(",", ""), errors="coerce") required_columns = [ "code", "sedol", "name", "conviction", "review_cutoff_date", "announcement_date", "effective_date", "event_name", "side", "flow_usdm", "shares_to_trade", "days_to_trade", ] missing = [c for c in required_columns if c not in out.columns] if missing: raise ValueError(f"PERISCOPE: vendor file missing required columns: {missing}") return out def compute_side(df): out = df.copy() out["side"] = pd.to_numeric(out["side"], errors="coerce") return out def coerce_event_types(df): out = df.copy() for c in ["review_cutoff_date", "announcement_date", "effective_date"]: if c in out.columns: out[c] = pd.to_datetime(out[c], errors="coerce").dt.date for c in ["flow_usdm", "shares_to_trade", "days_to_trade"]: if c in out.columns: out[c] = pd.to_numeric(out[c], errors="coerce") if "conviction" in out.columns: out["conviction"] = out["conviction"].astype(str).str.strip() if "name" in out.columns: out["name"] = out["name"].astype(str).str.strip() return out def _china_base_key(ric): s = str(ric).strip().upper() for suf in [".SZ", ".ZK", ".SH", ".SS"]: if s.endswith(suf): return s[:-3] return None def _match_key_from_ric(ric): s = str(ric).strip().upper() base = _china_base_key(s) return base if base is not None else s def get_model_universe_frame(): mfm = SignalMgr.getStatic("model_universe_frame") mfm = mfm.rename(columns={"RkdDisplayRIC": "internal_ric"}) return mfm def conviction_rank(x): s = str(x).strip().lower() if s == "high": return 4 if s == "medium": return 3 if s == "low": return 2 if s == "announced": return 1 return 0 def get_periscope_params(rebalConfig): min_dtt = getattr(rebalConfig, "periscopeMinDaysToTrade", 1.0) excluded_countries = getattr( rebalConfig, "periscopeExcludedCountries", [] ) if isinstance(excluded_countries, (list, tuple, set)): excluded_countries_set = set(str(x).strip().upper() for x in excluded_countries) else: excluded_countries_set = {str(excluded_countries).strip().upper()} PyLog.info(f"Excluded countries = {excluded_countries_set}") allowed = getattr( rebalConfig, "periscopeAllowedConvictions", ["High", "Medium", "Low", "Announced"], ) if isinstance(allowed, (list, tuple, set)): allowed_set = set(str(x).strip() for x in allowed) else: allowed_set = {str(allowed).strip()} # Rules: first match wins. DEFAULT must exist with pattern=None. # For FTSE/MSCI: end = cutoff + 3 days # Else: end = effective + 0 days rules = [ { "key": "FTSE", "pattern": r"\bFTSE\b", "trade_end_anchor": "cutoff", "trade_end_offset_days": 0, #"trade_end_anchor": "effective", #"trade_end_offset_days": 5, }, { "key": "MSCI", "pattern": r"\bMSCI\b", "trade_end_anchor": "cutoff", "trade_end_offset_days": 0, #"trade_end_anchor": "effective", #"trade_end_offset_days": 5, }, { "key": "DEFAULT", "pattern": None, "trade_end_anchor": "effective", "trade_end_offset_days": 0, #"trade_end_offset_days": 0, }, ] return { "min_dtt": float(min_dtt), "allowed_convictions": allowed_set, "event_rules": rules, "interest_window_days": int(getattr(rebalConfig, "periscopeInterestWindowDays", 360)), "excluded_countries": excluded_countries_set, } def standardize_sheet_columns(df): if df is None or df.empty: return df out = df.copy() cols = list(out.columns) def _move_after(col, after): nonlocal cols if col in cols and after in cols: cols.remove(col) cols.insert(cols.index(after) + 1, col) # ModelCntry right after internal_ric _move_after("ModelCntry", "internal_ric") # Change right after event_name _move_after("Change", "event_name") # Apply reordered columns out = out[cols] return out def write_meta_sheet(run, tradeYmd, wb=None, status="", note=""): rows = [ {"key": "tradeYmd", "value": tradeYmd}, {"key": "status", "value": status}, {"key": "note", "value": note}, ] if wb is None: rows += [ {"key": "periscope_enabled", "value": False}, {"key": "input_periscope_file", "value": ""}, {"key": "file_date_yyyymmdd", "value": ""}, {"key": "age_days", "value": ""}, {"key": "has_list_sheet", "value": False}, {"key": "has_change_sheet", "value": False}, ] else: rows += [ {"key": "periscope_enabled", "value": True}, {"key": "input_periscope_file", "value": wb.get("input_periscope_file", "")}, {"key": "file_date_yyyymmdd", "value": wb.get("file_date_yyyymmdd", "")}, {"key": "age_days", "value": wb.get("age_days", "")}, {"key": "has_list_sheet", "value": wb.get("df_list_raw") is not None and not wb.get("df_list_raw").empty}, {"key": "has_change_sheet", "value": wb.get("df_change_raw") is not None and not wb.get("df_change_raw").empty}, ] df_meta = pd.DataFrame(rows) write_dataframe_sheet(run, "meta", df_meta, "no meta") def write_periscope_detail_sheets(run, tradeYmd, wb): write_meta_sheet(run, tradeYmd, wb=wb, status="vendor_loaded", note="") def format_dates(df): df = df.copy() for col in ["Review Cut-off date", "Announcement Date", "Effective Date"]: if col in df.columns: df[col] = pd.to_datetime(df[col], errors="coerce").dt.date return df df_list = wb.get("df_list_raw") if df_list is not None: df_list = format_dates(df_list) write_dataframe_sheet( run, sheet_name="list_raw", df=df_list, note_if_empty="list sheet not found", ) df_change = wb.get("df_change_raw") if df_change is not None: df_change = format_dates(df_change) write_dataframe_sheet( run, sheet_name="change_raw", df=df_change, note_if_empty="change sheet not found", ) def write_dataframe_sheet(run, sheet_name, df, note_if_empty): w = run["writer"] sheet = sheet_name[:31] if df is None or df.empty: pd.DataFrame([{"note": note_if_empty}]).to_excel( w, sheet_name=sheet, index=False ) return df2 = df.copy() df2 = standardize_sheet_columns(df2) NON_NUMERIC_COLS = { "assetKey", "internal_ric", "external_ric", "code", "event_name", "conviction", "reason", "drop_reason", "ModelCntry", "Ticker", "tradeDate","restrictions_date", } for c in df2.columns: if c in NON_NUMERIC_COLS: continue if df2[c].dtype == object: s = df2[c].astype(str).str.strip() if s.eq("").all(): continue s2 = s.str.replace(",", "", regex=False) num = pd.to_numeric(s2, errors="coerce") if num.notna().mean() >= 0.80: df2[c] = num for c in df2.columns: if pd.api.types.is_datetime64_any_dtype(df2[c]): df2[c] = pd.to_datetime(df2[c], errors="coerce").dt.date # Also normalize date columns even if they are object/string mixed date_cols = [c for c in df2.columns if ("date" in str(c).lower()) ] for c in date_cols: if c in df2.columns: s = df2[c] # If column is object, it may contain datetime objects or strings with HH:MM:SS if s.dtype == object: # Convert anything parseable to datetime then drop time dt = pd.to_datetime(s, errors="coerce") if dt.notna().any(): df2[c] = dt.dt.date df2.to_excel(w, sheet_name=sheet, index=False) ws = w.sheets[sheet] for row in ws.iter_rows(min_row=2): for cell in row: col_name = ws.cell(row=1, column=cell.column).value if isinstance(cell.value, (int, float)): if col_name == "availableRate": cell.number_format = "0.00" else: cell.number_format = "#,##0" # Freeze first row ws.freeze_panes = "A2" # Apply auto filter to entire data range ws.auto_filter.ref = ws.dimensions from openpyxl.styles import Alignment if sheet == "process_logic": # Uniform width for documentation sheet uniform_width = 40 for col in ws.columns: col_letter = col[0].column_letter ws.column_dimensions[col_letter].width = uniform_width # Wrap text for all cells for row in ws.iter_rows(): for cell in row: cell.alignment = Alignment(wrap_text=True, vertical="top") ws.freeze_panes = "A2" else: # Original auto-width logic for all other sheets for col in ws.columns: max_len = 0 col_letter = col[0].column_letter for cell in col: if cell.value is not None: max_len = max(max_len, len(str(cell.value))) ws.column_dimensions[col_letter].width = max_len + 2 def write_active_events_sheet(run, df_governing_events): w = run["writer"] cols = PERISCOPE_EVENT_COLUMNS if df_governing_events is None or df_governing_events.empty: df_out = pd.DataFrame(columns=cols) else: keep = [c for c in cols if c in df_governing_events.columns] df_out = df_governing_events[keep].copy() if "assetKey" in df_out.columns: df_out = df_out.drop_duplicates(subset=["assetKey"], keep="first") write_dataframe_sheet(run, "active_events", df_out, "no active periscope events") def get_previous_details_path(today_details_path, tradeYmd): # tradeYmd is like "20251230" tradeDate = datetime.strptime(tradeYmd, "%Y%m%d").date() signalDate = PyDate.prevWeekday(tradeDate) prev = signalDate.strftime("%Y%m%d") return today_details_path.replace(tradeYmd, prev) def read_active_events(details_path): df = pd.read_excel(details_path, sheet_name="active_events") # Normalize date columns immediately after reading from Excel date_cols = [ "review_cutoff_date", "announcement_date", "effective_date", ] for c in date_cols: if c in df.columns: df[c] = pd.to_datetime(df[c], errors="coerce").dt.date if df is None or df.empty or "assetKey" not in df.columns: return pd.DataFrame() return df def write_optimizer_overrides_sheet(run, df_optimizer_overrides, df_unwind_effects): w = run["writer"] out = [] if df_optimizer_overrides is not None and not df_optimizer_overrides.empty: out.append(df_optimizer_overrides) if df_unwind_effects is not None and not df_unwind_effects.empty: out.append(df_unwind_effects) if not out: pd.DataFrame([{"note": "no overlay effects"}]).to_excel(w, sheet_name="optimizer_overrides", index=False) return df = pd.concat(out, axis=0, ignore_index=True) front = [c for c in PERISCOPE_EVENT_COLUMNS if c in df.columns] cols = list(df.columns) ordered = [c for c in front if c in cols] + [c for c in cols if c not in front] df = df[ordered] write_dataframe_sheet(run, "optimizer_overrides", df, "no overlay effects") return df def reorder_output_sheets(run, preferred_order, log=None): w = run.get("writer", None) if w is None: return # openpyxl excel wb = getattr(w, "book", None) if wb is None: return # current sheets current = [ws.title for ws in wb.worksheets] # build final order (keep only those that exist) wanted = [s for s in preferred_order if s in current] rest = [s for s in current if s not in wanted] final = wanted + rest # openpyxl: reorder by rearranging sheets in excel title_to_ws = {ws.title: ws for ws in wb.worksheets} wb._sheets = [title_to_ws[t] for t in final] if log is not None: log.info(f"PERISCOPE: reordered sheets -> {final}") def read_trade_restrictions(td: str) -> pd.DataFrame: """ Read daily trade restrictions. 1) Try trade date folder 2) If missing -> try signal_date (prev weekday) 3) If both missing -> log error and return empty dataframe """ signal_date = PyDate.prevWeekday(td).strftime("%Y%m%d") subdir = get_base_subdir_from_env() simdir_day = f"/signal/simulation/{subdir}/incoming/{td}" PyLog.info(f"{simdir_day}") if not os.path.exists(simdir_day): PyLog.info( f"Trade restrictions folder {simdir_day} does not exist for td={td}. " f"Checking signal_date={signal_date}" ) simdir_day = f"/signal/simulation/{subdir}/incoming/{signal_date}" PyLog.info(f"{simdir_day}") if not os.path.exists(simdir_day): PyLog.info( f"Restrictions folder missing for both td={td} and signal_date={signal_date}" ) return pd.DataFrame() # locate restriction file patterns = [ "universe_response*.csv", ] files = [] for p in patterns: files = sorted(glob.glob(os.path.join(simdir_day, p))) if files: break if not files: PyLog.info(f"No universe_response file found in {simdir_day}") return pd.DataFrame() path = files[0] df = pd.read_csv(path) df.columns = [str(c).strip().lower() for c in df.columns] # ensure expected columns exist for c in ["ric", "valid", "dnis", "dnil", "dnl", "dnh"]: if c not in df.columns: df[c] = np.nan # normalize booleans def _to_bool(s): if pd.isna(s): return False if isinstance(s, (bool, np.bool_)): return bool(s) x = str(s).strip().lower() return x in ("1", "true", "t", "y", "yes") for c in ["valid", "dnis", "dnil", "dnl", "dnh"]: df[c] = df[c].apply(_to_bool) df["external_ric"] = df["ric"].astype(str).str.strip() df["internal_ric"] = df["external_ric"].apply(RicX.externalric_to_ric) df["restrictions_date"] = simdir_day.split("/")[-1] df = df[df["valid"]].copy() return df def round_to_nearest_lot(qty: pd.Series, lot: pd.Series) -> pd.Series: """ Round ABS(qty) to nearest lot, preserve sign. If lot missing/<=0, keep original qty. """ q = pd.to_numeric(qty, errors="coerce") l = pd.to_numeric(lot, errors="coerce") out = q.copy() good = q.notna() & l.notna() & (l > 0) if good.any(): sgn = np.sign(out[good].astype(float)) absq = out[good].abs().astype(float) lotv = l[good].astype(float) rounded = np.round(absq / lotv) * lotv out.loc[good] = sgn * rounded return out def join_all_dataframes( df_optimizer_overrides: pd.DataFrame, trade_date_yyyymmdd: str, ): td = str(trade_date_yyyymmdd) signal_date = PyDate.prevWeekday(td) df_in = df_optimizer_overrides.copy() df_in.columns = [str(c).strip() for c in df_in.columns] df_in["assetKey"] = df_in["assetKey"].astype(str) df_in["side_num"] = pd.to_numeric(df_in.get("side"), errors="coerce") df_in["flow_usdm_num"] = pd.to_numeric(df_in.get("flow_usdm"), errors="coerce") required = ["assetKey", "side", "flow_usdm", "internal_ric"] missing = [c for c in required if c not in df_in.columns] if missing: raise ValueError(f"Input sheet '{sheet_name}' missing columns: {missing}") df_in = df_in.copy() df_in["assetKey"] = df_in["assetKey"].astype(str) df_in["side_num"] = pd.to_numeric(df_in["side"], errors="coerce") df_in["flow_usdm_num"] = pd.to_numeric(df_in["flow_usdm"], errors="coerce") # ----------------------- # Borrow availability (assetKey join) # ----------------------- PyLog.info(td) df_borrow = ShortAvailability.getRange(td, td, pbList=["GS"]) if df_borrow is None or df_borrow.empty: PyLog.info(f"Borrow data not available for {td}") df_borrow = pd.DataFrame() else: df_borrow = df_borrow.copy() df_borrow["assetKey"] = df_borrow["assetKey"].astype(str) df_borrow["availableQty"] = pd.to_numeric(df_borrow.get("availableQty"), errors="coerce") df_borrow["availableRate"] = pd.to_numeric(df_borrow.get("availableRate"), errors="coerce") df_borrow["notionalUSD"] = pd.to_numeric(df_borrow.get("notionalUSD"), errors="coerce") df_borrow = df_borrow.drop_duplicates(subset=["assetKey"], keep="last") # ----------------------- # Restrictions (internal_ric join) # ----------------------- df_restr = read_trade_restrictions(td) if df_restr is None or df_restr.empty: PyLog.info(f"Trade restrictions dataframe is empty for td={td}") df_restr = pd.DataFrame() # ----------------------- # Price frame latest (assetKey join) # ----------------------- pfm = SignalMgr.getSplitPriceFrame("price_frame_latest", signal_date) if pfm is None or pfm.empty: PyLog.info(f"Price frame not available for {signal_date}") pfm = pd.DataFrame() else: pfm = pfm.copy() pfm["assetKey"] = pfm["assetKey"].astype(str) pfm["priceCloseUSD"] = pd.to_numeric(pfm.get("priceCloseUSD"), errors="coerce") pfm = pfm.drop_duplicates(subset=["assetKey"], keep="last") # ----------------------- # Trading lot size (assetKey join) # ----------------------- lot = SignalMgr.getFrame("trading_lot_size", signal_date) if lot is None or lot.empty: PyLog.info(f"Trading lot size not available for {signal_date}") lot = pd.DataFrame() else: lot = lot.copy() lot["assetKey"] = lot["assetKey"].astype(str) lot["trading_lot_size"] = pd.to_numeric(lot.get("trading_lot_size"), errors="coerce") lot = lot.drop_duplicates(subset=["assetKey"], keep="last") # ----------------------- # Join everything # ----------------------- df = df_in.copy() # borrow if not df_borrow.empty: df = df.merge( df_borrow[["assetKey", "availableQty", "availableRate", "notionalUSD"]], on="assetKey", how="left", ) else: df["availableQty"] = np.nan df["availableRate"] = np.nan df["notionalUSD"] = np.nan # restrictions if not df_restr.empty: df = df.merge( df_restr[["internal_ric", "external_ric", "valid", "dnis", "dnil", "dnl", "dnh"]], on="internal_ric", how="left", ) else: df["external_ric"] = np.nan df["valid"] = False df["dnis"] = False df["dnil"] = False df["dnl"] = False df["dnh"] = False # price if not pfm.empty: df = df.merge( pfm[["assetKey", "priceCloseUSD"]], on="assetKey", how="left" ) else: df["priceCloseUSD"] = np.nan # lot if not lot.empty: df = df.merge( lot[["assetKey", "trading_lot_size"]], on="assetKey", how="left" ) else: df["trading_lot_size"] = np.nan # safety: drop any rows with missing assetKey df = df[df["assetKey"].notna()].copy() return df_in, df_borrow, df_restr, pfm, lot, df def cap_abs_flow_for_max_position(df: pd.DataFrame, periscope_trade_gross_usd: float, cap_usd: float) -> None: """ Modify df['abs_flow'] in-place so that no stock can receive more than cap_usd after allocation. Works separately for long and short sides to preserve neutrality. """ gross_side = float(periscope_trade_gross_usd) / 2.0 if gross_side <= 0: return max_weight = cap_usd / gross_side # ----- LONG SIDE ----- long_mask = df["side_num"] > 0 sum_long = df.loc[long_mask, "abs_flow"].sum() if sum_long > 0: max_flow_long = max_weight * sum_long cap_mask_long = long_mask & (df["abs_flow"] > max_flow_long) df.loc[cap_mask_long, "abs_flow"] = max_flow_long # ----- SHORT SIDE ----- short_mask = df["side_num"] < 0 sum_short = df.loc[short_mask, "abs_flow"].sum() if sum_short > 0: max_flow_short = max_weight * sum_short cap_mask_short = short_mask & (df["abs_flow"] > max_flow_short) df.loc[cap_mask_short, "abs_flow"] = max_flow_short def get_base_subdir_from_env(): import os import glob import re simconfig_name = os.environ.get("SIMCONFIG") if simconfig_name is None: raise ValueError("SIMCONFIG environment variable is not set") USER = os.environ.get("USER", "prod") config_path = f"/home/{USER}/gqr/model/model/trading/trading_configs/" cfg_files = glob.glob(f"{config_path}/*.cfg") target_file = None for f in cfg_files: if simconfig_name in f: target_file = f break if target_file is None: raise FileNotFoundError(f"{simconfig_name}.cfg not found in {config_path}") # Read file and extract subdir with open(target_file, "r") as fh: content = fh.read() match = re.search(r"subdir\s*=\s*['\"]([^'\"]+)['\"]", content) if not match: raise ValueError(f"subdir not found in {target_file}") return match.group(1) def build_manual_periscope_portfolio( trade_date_yyyymmdd: str, periscope_trade_gross_usd: float, max_short_utilization: float = 0.25, df_optimizer_overrides: pd.DataFrame = None, ): """ Build ONE clean output table: final_trades. """ td = str(trade_date_yyyymmdd) if df_optimizer_overrides is None: raise ValueError("df_optimizer_overrides is required") df_in, df_borrow, df_restr, pfm, lot, df = join_all_dataframes( df_optimizer_overrides=df_optimizer_overrides, trade_date_yyyymmdd=trade_date_yyyymmdd, ) df = df.copy() df["tradeDate"] = td # Ensure we do not carry fully blank trailing rows from Excel if "assetKey" in df.columns: df = df[df["assetKey"].notna()].copy() # ----------------------- # Base filtering + reasons # ----------------------- df["drop_reason"] = "" # Basic sanity bad_side = df["side_num"].isna() | (~df["side_num"].isin([1, -1])) df.loc[bad_side & (df["drop_reason"] == ""), "drop_reason"] = "BAD_SIDE" bad_flow = df["flow_usdm_num"].isna() | (df["flow_usdm_num"] == 0) df.loc[bad_flow & (df["drop_reason"] == ""), "drop_reason"] = "MISSING_OR_ZERO_FLOW" # Restrictions has_valid = df["valid"].fillna(False).astype(bool) df.loc[has_valid & df["dnh"].fillna(False) & (df["drop_reason"] == ""), "drop_reason"] = "RESTR_DNH" is_long = df["side_num"] > 0 is_short = df["side_num"] < 0 df.loc[has_valid & is_long & df["dnil"].fillna(False) & (df["drop_reason"] == ""), "drop_reason"] = "RESTR_DNIL" # dnl = do-not-liquidate (informational) => DO NOT drop on dnl df.loc[has_valid & is_short & df["dnis"].fillna(False) & (df["drop_reason"] == ""), "drop_reason"] = "RESTR_DNIS" # Price required df.loc[df["priceCloseUSD"].isna() & (df["drop_reason"] == ""), "drop_reason"] = "MISSING_PRICE" # Borrow required for shorts df.loc[is_short & df["availableQty"].isna() & (df["drop_reason"] == ""), "drop_reason"] = "MISSING_BORROW" # Borrow cap zero check (independent of intended qty) max_util = float(max_short_utilization) df["cap_qty"] = np.nan if max_util > 0: avail = pd.to_numeric(df["availableQty"], errors="coerce") cap_qty = (max_util * avail).astype("float64") cap_qty = cap_qty.where(is_short, np.nan) # only meaningful for shorts cap_qty = cap_qty.fillna(0.0).where(is_short, np.nan) # shorts missing/NaN => 0 cap df["cap_qty"] = cap_qty # cap <= 0 => drop (only shorts, only if not already dropped) cap_zero = is_short & (pd.to_numeric(df["cap_qty"], errors="coerce").fillna(0.0) <= 0.0) df.loc[cap_zero & (df["drop_reason"] == ""), "drop_reason"] = "BORROW_CAP_ZERO" # Keep flag (base) df["keep_for_trade"] = df["drop_reason"].eq("") # ----------------------- # Two-pass sizing for net-neutral + borrow-util drop # (1) allocate qty on current keep set # (2) drop shorts that exceed cap # (3) re-allocate on the remaining keep set to hit target gross # ----------------------- df["abs_flow"] = pd.to_numeric(df["flow_usdm_num"], errors="coerce").abs().fillna(0.0) #cap_abs_flow_for_max_position(df, periscope_trade_gross_usd, 500000.0) # Columns we will populate (keep rows only; others remain NaN) df["trade_value_usd"] = np.nan df["trade_qty_raw"] = np.nan df["trade_qty_after_borrow"] = np.nan df["trade_qty"] = np.nan df["trade_val_long_USD"] = np.nan df["trade_val_short_USD"] = np.nan gross_total = float(periscope_trade_gross_usd) gross_side = gross_total / 2.0 # net-neutral split def _allocate_trade_values(work_df: pd.DataFrame) -> pd.Series: """ Returns trade_value_usd allocated per side: 1) initial prorate by abs_flow 2) cap any position above 500K 3) re-prorate leftover across uncapped names once """ cap_usd = 500000.0 tv = pd.Series(0.0, index=work_df.index, dtype="float64") def _allocate_side(idx): if len(idx) == 0: return flows = work_df.loc[idx, "abs_flow"] s = float(flows.sum()) if s <= 0: return # ----- PASS 1 ----- tv_side = (flows / s) * gross_side # ----- CAP ----- capped = tv_side > cap_usd tv_side[capped] = cap_usd # ----- REALLOCATE LEFTOVER ----- remaining = gross_side - float(tv_side.sum()) if remaining > 0: uncapped_idx = tv_side.index[~capped] if len(uncapped_idx) > 0: flows_uncapped = flows.loc[uncapped_idx] s2 = float(flows_uncapped.sum()) if s2 > 0: add = (flows_uncapped / s2) * remaining tv_side.loc[uncapped_idx] += add tv.loc[idx] = tv_side # long side _allocate_side(work_df.index[work_df["side_num"] > 0]) # short side _allocate_side(work_df.index[work_df["side_num"] < 0]) return tv def _compute_qty(work_df: pd.DataFrame, trade_value_usd: pd.Series) -> pd.Series: """ trade_qty_raw = (trade_value_usd / priceCloseUSD) * side_num """ px = pd.to_numeric(work_df["priceCloseUSD"], errors="coerce") side = pd.to_numeric(work_df["side_num"], errors="coerce") qty = (trade_value_usd / px) * side return qty # Pass 1: allocate on current keep set keep_idx = df.index[df["keep_for_trade"]].tolist() if len(keep_idx) > 0: work = df.loc[keep_idx].copy() tv1 = _allocate_trade_values(work) q1 = _compute_qty(work, tv1) df.loc[work.index, "trade_value_usd"] = tv1 df.loc[work.index, "trade_qty_raw"] = q1 df.loc[work.index, "trade_qty_after_borrow"] = q1 # we do NOT cap; we drop if exceeded # Borrow-util drop for shorts (needs intended qty) if max_util > 0: cap = pd.to_numeric(df.loc[work.index, "cap_qty"], errors="coerce") cap = cap.where(df.loc[work.index, "side_num"] < 0, np.nan) exceed = (df.loc[work.index, "side_num"] < 0) & cap.notna() & (q1.abs().astype(float) > cap.astype(float)) df.loc[work.index[exceed] , "drop_reason"] = df.loc[work.index[exceed], "drop_reason"].mask( df.loc[work.index[exceed], "drop_reason"].eq(""), "BORROW_UTIL_EXCEEDED", ) # Recompute keep flag after borrow-util drop df["keep_for_trade"] = df["drop_reason"].eq("") # Pass 2: re-allocate on final keep set (so long/short totals hit gross_side) keep_idx2 = df.index[df["keep_for_trade"]].tolist() if len(keep_idx2) > 0: work2 = df.loc[keep_idx2].copy() tv2 = _allocate_trade_values(work2) q2 = _compute_qty(work2, tv2) df.loc[work2.index, "trade_value_usd"] = tv2 df.loc[work2.index, "trade_qty_raw"] = q2 df.loc[work2.index, "trade_qty_after_borrow"] = q2 # Round-lot df.loc[work2.index, "trade_qty"] = round_to_nearest_lot( df.loc[work2.index, "trade_qty_after_borrow"], df.loc[work2.index, "trading_lot_size"], ) # Round to zero => drop (but keep row) round_zero = df.loc[work2.index, "trade_qty"].fillna(0.0).astype(float).abs() == 0.0 df.loc[work2.index[round_zero] , "drop_reason"] = df.loc[work2.index[round_zero], "drop_reason"].mask( df.loc[work2.index[round_zero], "drop_reason"].eq(""), "ROUNDLOT_ZERO", ) df["keep_for_trade"] = df["drop_reason"].eq("") # Per-row long/short trade value columns (only for FINAL keep rows) final_keep_idx = df.index[df["keep_for_trade"]].tolist() if len(final_keep_idx) > 0: df.loc[final_keep_idx, "trade_val_long_USD"] = np.where( df.loc[final_keep_idx, "side_num"] > 0, df.loc[final_keep_idx, "trade_value_usd"], np.nan, ) df.loc[final_keep_idx, "trade_val_short_USD"] = np.where( df.loc[final_keep_idx, "side_num"] < 0, -df.loc[final_keep_idx, "trade_value_usd"], np.nan, ) # ----------------------- # Output formatting: ONE table with front cols first + rest # ----------------------- front_cols = [ "assetKey", "tradeDate", "internal_ric", "external_ric", "abs_flow", "side", "priceCloseUSD", "drop_reason", "keep_for_trade", "trade_value_usd", "trade_qty_raw", "cap_qty", "trade_qty_after_borrow", "trade_qty", "trade_val_long_USD", "trade_val_short_USD", "trading_lot_size", "availableQty", "availableRate", "notionalUSD", "valid", "dnis", "dnil", "dnl", "dnh", ] cols_front = [c for c in front_cols if c in df.columns] cols_rest = [c for c in df.columns if c not in cols_front] df_trades = df[cols_front + cols_rest].copy().reset_index(drop=True) usd_cols = ["trade_value_usd", "trade_val_long_USD", "trade_val_short_USD" ] for c in usd_cols: if c in df_trades.columns: df_trades[c] = pd.to_numeric(df_trades[c], errors="coerce").round(0) # ----------------------- # Summary row # ----------------------- summary = {c: np.nan for c in df_trades.columns} summary["assetKey"] = "TOTAL" summary["tradeDate"] = td summary["abs_flow"] = float(pd.to_numeric(df_trades["abs_flow"], errors="coerce").fillna(0.0).sum()) tvl = pd.to_numeric(df_trades.get("trade_val_long_USD"), errors="coerce").fillna(0.0) tvs = pd.to_numeric(df_trades.get("trade_val_short_USD"), errors="coerce").fillna(0.0) summary["trade_val_long_USD"] = float(tvl.sum()) summary["trade_val_short_USD"] = float(tvs.sum()) df_trades = pd.concat([df_trades, pd.DataFrame([summary])], ignore_index=True) return { "input": df_in, "borrow": df_borrow, "restrictions": df_restr, "final_trades": df_trades, } # the final trades sheeet created in apply_periscope_overlay is actually manually updated and fine tuned # and that serves as the input for the below function. This function then creates the manual periscope target file # this function can be removed once we have periscope integrated into our base strategy optimizer or into a # standalone optimizer def build_final_periscope_targets(df_manual: pd.DataFrame, signal_date): UNKNOWN = "UNKNOWN" PRICE_COL_OUT = f"price_close_{signal_date.strftime('%Y%m%d')}" # ----------------------- # Detect input format # ----------------------- cols = set(df_manual.columns) if {"internal_ric", "periscope_qty"}.issubset(cols): df = df_manual.copy() PyLog.info("[periscope_targets] using INTERNAL format") elif {"ric", "targetposition"}.issubset(cols): PyLog.info("[periscope_targets] using EXTERNAL format → converting") df = df_manual.copy() def safe_internal_ric(external_ric): if pd.isna(external_ric): return UNKNOWN try: out = RicX.externalric_to_ric(str(external_ric).strip()) return out if out else UNKNOWN except Exception: return UNKNOWN df["internal_ric"] = df["ric"].apply(safe_internal_ric) df["periscope_qty"] = pd.to_numeric(df["targetposition"], errors="coerce") else: raise RuntimeError( "Input must contain either " "{internal_ric, periscope_qty} OR {ric, targetposition}" ) # ----------------------- # Normalize # ----------------------- df["internal_ric"] = df["internal_ric"].astype(str).str.strip() input_qty_sum = pd.to_numeric(df["periscope_qty"], errors="coerce").sum() # ----------------------- # Model universe (assetKey) # ----------------------- PyLog.info("[periscope_targets] loading model_universe_frame") mfm = SignalMgr.getStatic("model_universe_frame") if mfm is None or len(mfm) == 0: raise RuntimeError("model_universe_frame is empty") mfm = mfm.copy() from datetime import datetime # Normalize signal_date if isinstance(signal_date, str): signalDate_dt = datetime.strptime(signal_date, "%Y%m%d").date() else: signalDate_dt = signal_date # 🔥 CRITICAL: filter universe by date (MISSING IN YOUR CODE) mfm = mfm[ (mfm['startDate'] <= signalDate_dt) & ( mfm['endDate'].isna() | (mfm['endDate'] >= signalDate_dt) ) ] # Defensive universe filter try: model_countries = Universe.getModelCountries("univ_asia_v1") mfm = mfm[mfm["ModelCntry"].isin(model_countries)] except Exception: PyLog.info("[periscope_targets] universe filter skipped") mfm["assetKey"] = mfm["assetKey"].astype(str).str.strip() mfm["RkdDisplayRIC"] = mfm["RkdDisplayRIC"].astype(str).str.strip() mfm = mfm[["assetKey", "RkdDisplayRIC"]] df = df.merge( mfm.rename(columns={"RkdDisplayRIC": "internal_ric_mfm"}), left_on="internal_ric", right_on="internal_ric_mfm", how="left", ) df["assetKey"] = df["assetKey"].fillna(UNKNOWN) df.drop(columns=["internal_ric_mfm"], inplace=True) # ----------------------- # External ric (always output) # ----------------------- def safe_external_ric(internal_ric): if pd.isna(internal_ric): return UNKNOWN try: out = RicX.ric_to_externalric(str(internal_ric).strip()) return out if out else UNKNOWN except Exception: return UNKNOWN df["ric"] = df["internal_ric"].apply(safe_external_ric) # ----------------------- # Position # ----------------------- df["position"] = 0 # ----------------------- # Price lookup # ----------------------- PyLog.info(f"[periscope_targets] loading price_frame_latest for {signal_date}") pfm = SignalMgr.getSplitPriceFrame("price_frame_latest", signal_date) if pfm is not None and not pfm.empty: pfm = pfm.copy() pfm["assetKey"] = pfm["assetKey"].astype(str).str.strip() pfm["priceCloseUSD"] = pd.to_numeric(pfm.get("priceCloseUSD"), errors="coerce") pfm = pfm.drop_duplicates(subset=["assetKey"], keep="last") pfm = pfm[["assetKey", "priceCloseUSD"]] df = df.merge(pfm, on="assetKey", how="left") else: df["priceCloseUSD"] = np.nan df[PRICE_COL_OUT] = df["priceCloseUSD"].where( df["priceCloseUSD"].notna(), UNKNOWN ) df.drop(columns=["priceCloseUSD"], inplace=True) # ----------------------- # Cleanup # ----------------------- df["assetKey"] = df["assetKey"].fillna(UNKNOWN).astype(str) df["ric"] = df["ric"].fillna(UNKNOWN).astype(str) df["internal_ric"] = df["internal_ric"].fillna(UNKNOWN).astype(str) df["periscope_qty"] = pd.to_numeric(df["periscope_qty"], errors="coerce") # ----------------------- # Output # ----------------------- df_out = df[ [ "assetKey", "ric", "internal_ric", "position", "periscope_qty", PRICE_COL_OUT, ] ].copy() # ----------------------- # Quantity check (CRITICAL) # ----------------------- output_qty_sum = pd.to_numeric(df_out["periscope_qty"], errors="coerce").sum() if not np.isclose(input_qty_sum, output_qty_sum, atol=1e-6): raise RuntimeError( f"Quantity mismatch! input={input_qty_sum}, output={output_qty_sum}" ) PyLog.info( f"[periscope_targets] quantity check passed: " f"input_sum={input_qty_sum}, output_sum={output_qty_sum}" ) # ----------------------- # Summary stats (IMPORTANT) # ----------------------- input_rows = len(df) output_rows = len(df_out) missing_assetKey = (df_out["assetKey"] == "UNKNOWN").sum() missing_ric = (df_out["ric"] == "UNKNOWN").sum() price_col = PRICE_COL_OUT missing_price = (df_out[price_col] == "UNKNOWN").sum() PyLog.info( f"[periscope_targets] summary: " f"input_rows={input_rows}, " f"output_rows={output_rows}, " f"missing_assetKey={missing_assetKey}, " f"missing_ric={missing_ric}, " f"missing_price={missing_price}" ) return df_out ### Entry point to create final manual target file def run_periscope_manual_target_build(input_file: str, output_file: str, signal_date): PyLog.info(f"[periscope_targets] reading manual sheet from: {input_file}") try: if input_file.endswith(".csv"): df_manual = pd.read_csv(input_file) else: df_manual = pd.read_excel(input_file, sheet_name="final_manual_trades") except Exception as e: raise RuntimeError( f"Failed to read manual input from {input_file}: {e}" ) if df_manual.empty: raise RuntimeError(f"final_manual_trades sheet is empty in {input_file}") df_manual.columns = df_manual.columns.str.strip() # Build targets df_targets = build_final_periscope_targets( df_manual=df_manual, signal_date=signal_date, ) PyLog.info(f"[periscope_targets] writing output file: {output_file} rows={len(df_targets)}") # Write output df_targets.to_csv(output_file, index=False) PyLog.info( f"[periscope_targets] wrote final targets: {output_file}, rows={len(df_targets)}" ) return df_targets def main(): import argparse from datetime import datetime parser = argparse.ArgumentParser(description="Periscope manual target builder") parser.add_argument( "--input", help="Manual input file from colleague (Excel/CSV)", ) parser.add_argument( "--output", default="final_targets.csv", help="Output CSV file", ) parser.add_argument( "--signal-date", help="Signal date in YYYYMMDD", ) args = parser.parse_args() # If no manual file provided → do nothing if not args.manual_file: print("No --input provided. Exiting.") return if not args.signal_date: raise ValueError("--signal-date is required when using --input") signal_date = datetime.strptime(args.signal_date, "%Y%m%d").date() # Run builder run_periscope_manual_target_build( input_file=args.manual_file, output_file=args.output, signal_date=signal_date, ) def main(): import argparse from datetime import datetime parser = argparse.ArgumentParser() parser.add_argument("--input", required=True) parser.add_argument("--output", required=True) parser.add_argument("--signal-date", required=True) args = parser.parse_args() signal_date = datetime.strptime(args.signal_date, "%Y%m%d").date() run_periscope_manual_target_build( input_file=args.input, output_file=args.output, signal_date=signal_date, ) if __name__ == "__main__": main()