"""Corporate-action policy (docs/DATA_CARD.md §adjustments).

- Splits: applied explicitly from the instrument master (`/instruments/{symbol}`: action_type=split, ex_date, value);
  for a bar in session d < ex_date, prices /= ratio and volume *= ratio for every split with ex_date > d. This equals the
  server-side `adjustment=split` series (verified in tests/integration/test_live_api.py on real rows).
- Cash dividends are NOT applied to intraday returns: a dividend adjustment rescales all history before the ex-date and
  therefore makes any windowed return depend on the end of the download range (window-dependent artifact). Intraday
  log returns within a session are invariant to the dividend factor; the only affected quantity is the overnight gap
  feature across an ex-dividend date, which we leave raw and flag.
- Mergers / spin-offs: not available from the vendor; a symbol disappears (delisted suffix) and the series ends.
"""
from __future__ import annotations

import numpy as np
import pandas as pd
import polars as pl


def split_factor_by_session(session_dates: np.ndarray, splits: list[dict]) -> np.ndarray:
    """Cumulative split ratio applying to each session date (product of ratios with ex_date > session)."""
    sd = pd.to_datetime(pd.Series(session_dates)).to_numpy().astype("datetime64[D]")
    factor = np.ones(len(sd), dtype=np.float64)
    for s in splits:
        if s.get("action_type") != "split":
            continue
        ex = np.datetime64(str(s["ex_date"]), "D")
        ratio = float(s["value"])
        if ratio <= 0:
            continue
        factor[sd < ex] *= ratio
    return factor


def apply_splits(bars: pl.DataFrame, splits: list[dict]) -> pl.DataFrame:
    """Return split-adjusted copy (raw columns kept as *_raw)."""
    f = split_factor_by_session(bars["session_date"].to_numpy(), splits)
    out = bars.with_columns(pl.Series("split_factor", f))
    return out.with_columns([
        (pl.col(c) / pl.col("split_factor")).alias(c) for c in ("open", "high", "low", "close")
    ] + [(pl.col("volume") * pl.col("split_factor")).alias("volume")]).with_columns(
        pl.lit("split").alias("adjustment_type"))


def ex_dividend_sessions(actions: list[dict]) -> set[str]:
    return {str(a["ex_date"]) for a in actions if a.get("action_type") == "dividend"}