"""Finance / CFO report schemas — wraps finance_service outputs.""" from __future__ import annotations from typing import Any, Optional from pydantic import BaseModel # ── Lost sales ───────────────────────────────────────────────────────── class LostSalesRow(BaseModel): sku: str name: Optional[str] = None tier: Optional[str] = None category: Optional[str] = None supplier: Optional[str] = None stock_now: float wh_stock: float store_stock: float total_incoming_h: float total_demand_h: float stockout_week: str weeks_out_of_stock: int lost_demand_units: float cost_price: float selling_price: float lost_sales_cost_eur: float lost_sales_revenue_eur: float class CategoryAgg(BaseModel): category: str skus_at_risk: int = 0 lost_units: float = 0 lost_cost_eur: float = 0 lost_revenue_eur: float = 0 class TierAgg(BaseModel): tier: str skus_at_risk: int = 0 lost_units: float = 0 lost_cost_eur: float = 0 lost_revenue_eur: float = 0 class LostSalesSummary(BaseModel): total_cost_eur: float total_revenue_eur: float total_units: float n_total_skus: int n_at_risk: int n_safe: int n_new_excluded: int = 0 class LostSalesResponse(BaseModel): summary: LostSalesSummary rows: list[LostSalesRow] by_category: list[CategoryAgg] by_tier: list[TierAgg] top5: list[dict] # ── Lost sales — MONTHLY (past-looking, RUC-focused) ────────────────── class LostSalesMonthlyRow(BaseModel): product_id: int sku: str name: Optional[str] = None category: Optional[str] = None tier: Optional[str] = None expected_qty: float actual_qty: float lost_qty: float lost_ruc_eur: float ruc_rate: float n_oos_weeks: int oos_weeks: str class LostSalesMonthlyTierAgg(BaseModel): tier: str n_skus: int lost_qty: float lost_ruc_eur: float class LostSalesMonthlyCategoryAgg(BaseModel): category: str n_skus: int lost_qty: float lost_ruc_eur: float class LostSalesMonthlySummary(BaseModel): year: int month: int label: str weeks_covered: str tiers: list[str] total_lost_ruc_eur: float total_lost_qty: float n_skus_lost: int n_skus_evaluated: int class LostSalesMonthlyResponse(BaseModel): summary: LostSalesMonthlySummary rows: list[LostSalesMonthlyRow] by_tier: list[LostSalesMonthlyTierAgg] by_category: list[LostSalesMonthlyCategoryAgg] # ── Locked cash ──────────────────────────────────────────────────────── class LockedCashRow(BaseModel): sku: str name: Optional[str] = None tier: Optional[str] = None category: Optional[str] = None supplier: Optional[str] = None stock_now: float wh_stock: float store_stock: float weekly_demand: float weeks_cover: Optional[float] = None effective_weeks_cover: Optional[float] = None incoming_in_lt: float = 0.0 status: str = "OVERSTOCK" lead_time_weeks: float target_stock_weeks: float target_stock_units: float excess_units: float cost_price: float selling_price: float stock_value_eur: float locked_cash_cost_eur: float locked_cash_revenue_eur: float class LockedCashCategoryAgg(BaseModel): category: str sku_count: int total_stock_eur: float total_excess_eur: float excess_pct: float class LockedCashSupplierAgg(BaseModel): supplier: str sku_count: int total_excess_eur: float class LockedCashSummary(BaseModel): total_stock_eur: float total_excess_cost_eur: float total_excess_rev_eur: float n_skus: int n_overstocked: int n_new_excluded: int = 0 n_below_threshold: int = 0 n_missing_lt: int = 0 class LockedCashResponse(BaseModel): summary: LockedCashSummary rows: list[LockedCashRow] by_category: list[LockedCashCategoryAgg] by_supplier: list[LockedCashSupplierAgg] # ── Contest risk ─────────────────────────────────────────────────────── class ContestRiskRow(BaseModel): sku: str name: Optional[str] = None tier: Optional[str] = None category: Optional[str] = None supplier: Optional[str] = None stock_now: float wh_stock: float = 0 store_stock: float = 0 incoming_before_contest: float available_at_contest: float incoming_during_contest: float baseline_demand_july: float promo_uplift_factor: float contest_demand_july: float stockout_baseline: str stockout_contest: str gap_units: float gap_cost_eur: float gap_revenue_eur: float # First contest week the SKU is projected to run dry (week-by-week # running balance), e.g. "CW26". None when it never stocks out. first_stockout_cw: Optional[str] = None contest_month_key: Optional[int] = None contest_section: Optional[str] = None contest_parent: Optional[str] = None class ContestSummary(BaseModel): uplift_factor: float n_skus_evaluated: int n_at_risk_baseline: int n_at_risk_contest: int gap_revenue_eur_baseline: float gap_revenue_eur_contest: float gap_cost_eur_contest: float n_new_excluded: int = 0 contest_month_key: Optional[int] = None contest_weeks: list[str] = [] class ContestRiskResponse(BaseModel): summary: ContestSummary rows: list[ContestRiskRow] top10_at_risk: list[dict] # ── Slow movers ──────────────────────────────────────────────────────── class SlowMoverRow(BaseModel): sku: str name: Optional[str] = None tier: Optional[str] = None category: Optional[str] = None supplier: Optional[str] = None stock_units: float stock_cost_eur: float stock_revenue_eur: float weekly_demand: float weeks_cover: Optional[float] = None lead_time_weeks: Optional[float] = None classification: str excess_units: float excess_cost_eur: float excess_revenue_eur: float class SlowMoverSummary(BaseModel): dead_stock_eur: float slow_mover_eur: float overstock_eur: float healthy_eur: float understock_eur: float new_listing_eur: float = 0 gadgets_eur: float non_food_eur: float n_skus: int by_class_count: dict[str, int] class SlowMoverGroupRow(BaseModel): group_value: str DEAD_STOCK: float = 0 SLOW_MOVER: float = 0 OVERSTOCK: float = 0 HEALTHY: float = 0 UNDERSTOCK: float = 0 NEW_LISTING: float = 0 total_stock_eur: float = 0 unhealthy_eur: float = 0 unhealthy_pct: float = 0 class SlowMoverResponse(BaseModel): summary: SlowMoverSummary rows: list[SlowMoverRow] by_category: list[SlowMoverGroupRow] by_supplier: list[SlowMoverGroupRow] # ── Bridge ───────────────────────────────────────────────────────────── class BridgeMonthRow(BaseModel): """Per-month bridge summary (for the monthly bridge endpoint). Mirrors what the UI expects per month.""" month: str month_key: int tier: str n_skus: int days_observed: float days_in_month: int coverage_pct: float plan_margin_total: float actual_margin_total: float total_variance: float volume_effect: float price_effect: float mix_effect: float cogs_effect: float class BridgeRow(BaseModel): sku: str tier: Optional[str] = None category: Optional[str] = None period: str plan_qty: Optional[float] = None actual_qty: Optional[float] = None plan_price: Optional[float] = None actual_unit_price: Optional[float] = None plan_revenue: Optional[float] = None actual_revenue: Optional[float] = None volume_effect_eur: float price_effect_eur: float mix_effect_eur: float plan_cogs: Optional[float] = None actual_cogs: Optional[float] = None cogs_effect_eur: float margin_effect_total_eur: float class BridgeCategoryRow(BaseModel): category: str volume_effect: float price_effect: float mix_effect: float cogs_effect: float total_margin_effect: float class BridgeWaterfallBar(BaseModel): label: str value: float type: str # base | increase | decrease | total class BridgeSummary(BaseModel): window_weeks: list[int] = [] n_skus: int = 0 plan_margin: float = 0 actual_margin: float = 0 volume_effect: float = 0 price_effect: float = 0 mix_effect: float = 0 cogs_effect: float = 0 total_variance: float = 0 class BridgeResponse(BaseModel): available: bool type: str summary: BridgeSummary rows: list[BridgeRow] by_category: list[BridgeCategoryRow] waterfall: list[BridgeWaterfallBar] # ── Dashboard ────────────────────────────────────────────────────────── class FinanceDashboard(BaseModel): anchor_year_week: str horizon_weeks: int n_products_in_scope: int n_dormant_excluded: int = 0 lost_sales: LostSalesSummary locked_cash: LockedCashSummary contest_risk: ContestSummary slow_movers: SlowMoverSummary bridge: dict top5_lost_sales: list[dict] top10_contest: list[dict]