"""Splits and cash dividends (FinData `corporate_actions`, vendor meta files) -> per-symbol ex-date adjustment table.

The extract is produced once on the storage node (read-only `psql \\copy`, see docs/API_INTEGRATION.md) because the API
has no corporate-actions route; its SHA-256 is in artifacts/manifests/corporate_actions.json. Raw bars stay unadjusted;
adjustments are applied explicitly in features.daily_block (ex-date rule).
"""
from __future__ import annotations

from pathlib import Path

import polars as pl


def load_actions(path: Path) -> pl.DataFrame:
    """Raw Postgres copy: id, instrument_id, action_type (split|dividend), ex_date, value, ... .
    Vendor quirk: dividends are keyed `firstrate:stock:<SYM>_DIVS`; `value` = split ratio (new per old) or cash amount."""
    if not path.exists():
        return pl.DataFrame(schema={"symbol": pl.Utf8, "ex_date": pl.Date, "split_ratio": pl.Float64, "dividend": pl.Float64})
    a = pl.read_csv(path, try_parse_dates=True, infer_schema_length=10000)
    a = a.with_columns([
        pl.col("instrument_id").str.replace("^firstrate:[a-z]+:", "").str.replace("_DIVS$", "").alias("symbol"),
        pl.col("ex_date").cast(pl.Date), pl.col("value").cast(pl.Float64),
    ])
    splits = a.filter(pl.col("action_type") == "split").group_by(["symbol", "ex_date"]).agg(pl.col("value").product().alias("split_ratio"))
    divs = a.filter(pl.col("action_type") == "dividend").group_by(["symbol", "ex_date"]).agg(pl.col("value").sum().alias("dividend"))
    out = splits.join(divs, on=["symbol", "ex_date"], how="full", coalesce=True).with_columns([
        pl.col("split_ratio").fill_null(1.0), pl.col("dividend").fill_null(0.0)])
    out = out.filter((pl.col("split_ratio") > 0) & (pl.col("dividend") >= 0))
    return out.sort(["symbol", "ex_date"])


def split_events_in_window(actions: pl.DataFrame, symbol: str, start, end) -> int:
    if actions.height == 0:
        return 0
    return actions.filter((pl.col("symbol") == symbol) & (pl.col("ex_date") >= start) & (pl.col("ex_date") <= end)
                          & (pl.col("split_ratio") != 1.0)).height