from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path

import duckdb
import pandas as pd

from .config import Settings


FACTOR_DATASET_VERSION = "0.1.0"


@dataclass(frozen=True)
class FactorDatasetResult:
    rows: int
    stocks: int
    periods: int
    output_path: Path


def build_monthly_factor_panel(
    settings: Settings,
    start_date: str = "2019-01-01",
    end_date: str | None = None,
) -> FactorDatasetResult:
    """Build a point-in-time monthly factor panel from the M1 data assets.

    Financial rows only become visible on ``available_trade_date``. Forward
    returns are labels and are never used to construct the factor columns.
    Industry membership is the current M1 taxonomy, so historical industry
    drift remains an explicit data limitation rather than being hidden.
    """
    price_path = str(settings.daily_price_path).replace("'", "''")
    finance_path = str(settings.financial_pit_path).replace("'", "''")
    master_path = str(settings.processed_dir / "security_master.parquet").replace("'", "''")
    end_filter = ""
    if end_date:
        safe_end = pd.Timestamp(end_date).date().isoformat()
        end_filter = f"AND trade_date <= DATE '{safe_end}'"
    safe_start = pd.Timestamp(start_date).date().isoformat()

    query = f"""
    WITH monthly_raw AS (
      SELECT source_code,
             date_trunc('month', trade_date) AS month,
             last_day(date_trunc('month', trade_date)) AS rebalance_date,
             arg_max(trade_date, trade_date) AS price_date,
             arg_max(close_qfq, trade_date) AS close_qfq,
             arg_max(close_raw, trade_date) AS close_raw,
             avg(amount_yuan) AS avg_amount
      FROM read_parquet('{price_path}')
      WHERE trade_date >= DATE '{safe_start}' {end_filter}
        AND close_qfq > 0
      GROUP BY source_code, date_trunc('month', trade_date)
    ), monthly AS (
      SELECT *,
             lead(price_date) OVER (PARTITION BY source_code ORDER BY month) AS next_date,
             lead(close_qfq) OVER (PARTITION BY source_code ORDER BY month) / close_qfq - 1 AS forward_return
      FROM monthly_raw
    ), joined AS (
      SELECT m.*, f.report_period, f.available_trade_date,
             f.weighted_roe_reported, f.gross_margin_reported,
             f.operating_revenue_yoy_reported, f.net_profit_yoy_reported,
             f.debt_ratio_reported, f.current_ratio_reported,
             f.asset_turnover_reported, f.ocf_net_profit_ratio_reported,
             f.research_development_expense_ytd, f.operating_revenue_ytd,
             f.parent_equity, f.total_shares,
             row_number() OVER (
               PARTITION BY m.source_code, m.rebalance_date
               ORDER BY f.available_trade_date DESC, f.report_period DESC
             ) AS finance_rank
      FROM monthly m
      LEFT JOIN read_parquet('{finance_path}') f
        ON f.source_code = m.source_code
       AND f.available_trade_date <= m.price_date
    )
    SELECT j.source_code, s.security_id, s.name,
           s.industry_l1_code, s.industry_l1_name,
           s.industry_l2_code, s.industry_l2_name,
           j.rebalance_date, j.price_date, j.next_date, j.report_period, j.available_trade_date,
           j.close_qfq AS close, j.close_raw, j.avg_amount, j.forward_return,
           j.weighted_roe_reported AS profitability,
           j.gross_margin_reported AS gross_margin,
           (j.operating_revenue_yoy_reported + j.net_profit_yoy_reported) / 2 AS growth,
           -j.debt_ratio_reported AS balance_sheet,
           j.current_ratio_reported AS liquidity,
           j.asset_turnover_reported AS efficiency,
           j.ocf_net_profit_ratio_reported AS cash_quality,
           CASE WHEN j.operating_revenue_ytd > 0
                THEN j.research_development_expense_ytd / j.operating_revenue_ytd END AS rd_intensity,
           CASE WHEN j.parent_equity > 0 AND j.total_shares > 0 AND j.close_raw > 0
                THEN j.parent_equity / (j.close_raw * j.total_shares) END AS book_to_price,
           '{FACTOR_DATASET_VERSION}' AS dataset_version
    FROM joined j
    LEFT JOIN read_parquet('{master_path}') s USING (source_code)
    WHERE j.finance_rank = 1 AND j.forward_return IS NOT NULL
      AND j.available_trade_date IS NOT NULL
    ORDER BY j.rebalance_date, j.source_code
    """
    with duckdb.connect() as connection:
        panel = connection.execute(query).fetchdf()

    for column in ["rebalance_date", "price_date", "next_date", "report_period", "available_trade_date"]:
        panel[column] = pd.to_datetime(panel[column])
    if not panel.empty and (panel["available_trade_date"] > panel["price_date"]).any():
        raise AssertionError("factor panel contains financial information unavailable at rebalance")

    settings.processed_dir.mkdir(parents=True, exist_ok=True)
    output_path = settings.processed_dir / "factor_monthly_panel.parquet"
    panel.to_parquet(output_path, index=False)
    panel.to_csv(settings.processed_dir / "factor_monthly_panel.csv", index=False, encoding="utf-8-sig")
    return FactorDatasetResult(
        rows=len(panel),
        stocks=int(panel["source_code"].nunique()),
        periods=int(panel["rebalance_date"].nunique()),
        output_path=output_path,
    )