"""Pydantic response models for the promo module. Three lenses on promo data: 1. History — every (SKU × contiguous-run) of erp_promo_weeks 2. Performance — same events but with before/during/after uplift math computed from v_sales_weekly_full 3. Proposals — writes into promo_proposals (Planner submits; MyProposals lists) The performance math mirrors `build_promo_performance.py` verbatim so the unified app gives the same numbers Streamlit shows today. """ from __future__ import annotations from typing import Optional from pydantic import BaseModel # ---------------------------------------------------------------------- # Promo history — one row per contiguous promo run per SKU # ---------------------------------------------------------------------- class PromoEvent(BaseModel): sku: str name: Optional[str] = None category: Optional[str] = None tier: Optional[str] = None start_year: int start_week: int end_year: int end_week: int n_weeks: int promo_types: Optional[str] = None cw_label: str # "CW10/2025 → CW14/2025" class PromoHistoryResponse(BaseModel): events: list[PromoEvent] n_events: int n_skus: int note: Optional[str] = None # ---------------------------------------------------------------------- # Promo performance — uplift + cannibalization per event # ---------------------------------------------------------------------- class PromoPerformance(BaseModel): sku: str name: Optional[str] = None category: Optional[str] = None tier: Optional[str] = None start_year: int start_week: int end_year: int end_week: int n_weeks: int promo_types: Optional[str] = None cw_label: str qty_before_avg: Optional[float] = None qty_during_avg: Optional[float] = None qty_after_avg: Optional[float] = None actual_uplift: Optional[float] = None # during / before cannibalization: Optional[float] = None # after / before net_effect: Optional[float] = None # (during+after) / (before × n_weeks) class PromoCategoryAgg(BaseModel): category: str n_events: int median_uplift: Optional[float] = None median_cannibalization: Optional[float] = None median_net_effect: Optional[float] = None class PromoPerformanceResponse(BaseModel): events: list[PromoPerformance] by_category: list[PromoCategoryAgg] n_events: int n_skus: int n_with_uplift: int median_uplift: Optional[float] = None median_cannibalization: Optional[float] = None median_net_effect: Optional[float] = None max_weeks_filter: int note: Optional[str] = None # ---------------------------------------------------------------------- # Promo proposals — read + write # ---------------------------------------------------------------------- class PromoProposalSkuLine(BaseModel): """One SKU within a proposal. Lives in the `skus` JSONB column.""" sku: str name: Optional[str] = None forecast_qty: Optional[float] = None # planner-typed forecast analog_uplift: Optional[float] = None # what the analog forecast suggested class PromoProposal(BaseModel): id: int name: Optional[str] = None source: Optional[str] = None mechanic: Optional[str] = None discount_pct: Optional[float] = None start_year: Optional[int] = None start_week: Optional[int] = None end_year: Optional[int] = None end_week: Optional[int] = None skus: list[PromoProposalSkuLine] = [] n_skus: int = 0 status: str objective: Optional[str] = None # "logika akcije": margin|traffic|stock_clear|acquisition proposed_by_id: Optional[int] = None proposed_by: Optional[str] = None channels: Optional[str] = None log: Optional[str] = None planner_state: Optional[dict] = None # full Planner working copy (reopen/edit) created_at: Optional[str] = None cw_label: Optional[str] = None # "CW22/2026 → CW25/2026" class PromoProposalCreate(BaseModel): name: str mechanic: str # "Discount %" / "1+1" / "2+1" / "3+1" / ... discount_pct: Optional[float] = None start_year: int start_week: int end_year: int end_week: int skus: list[PromoProposalSkuLine] channels: Optional[str] = "retail" status: str = "draft" objective: Optional[str] = None # "logika akcije": margin|traffic|stock_clear|acquisition log: Optional[str] = None source: Optional[str] = "planner" planner_state: Optional[dict] = None # full Planner working copy (reopen/edit) proposed_by_id: Optional[int] = None # filled by auth later class PromoProposalUpdate(BaseModel): """Overwrite an existing proposal's editable content. All fields optional so a partial re-save (e.g. status-only) leaves the rest intact. `log_line` is appended to the audit trail rather than replacing `log`.""" name: Optional[str] = None mechanic: Optional[str] = None discount_pct: Optional[float] = None start_year: Optional[int] = None start_week: Optional[int] = None end_year: Optional[int] = None end_week: Optional[int] = None skus: Optional[list[PromoProposalSkuLine]] = None channels: Optional[str] = None status: Optional[str] = None objective: Optional[str] = None # "logika akcije": margin|traffic|stock_clear|acquisition log_line: Optional[str] = None planner_state: Optional[dict] = None class PromoProposalListResponse(BaseModel): proposals: list[PromoProposal] n_total: int by_status: dict[str, int] = {} # draft / submitted / approved / rejected / revision note: Optional[str] = None # ---------------------------------------------------------------------- # Analog forecast — used by Planner Step 3 # ---------------------------------------------------------------------- class AnalogSkuHit(BaseModel): """One historical promo on the same SKU used to seed the forecast.""" start_year: int start_week: int end_year: int end_week: int cw_label: str n_weeks: int promo_types: Optional[str] = None discount_pct: Optional[float] = None qty_before_avg: Optional[float] = None qty_during_avg: Optional[float] = None actual_uplift: Optional[float] = None similarity: float # 0..1 weight used in the aggregate class AnalogForecast(BaseModel): sku: str name: Optional[str] = None n_analogs: int # historical promos used median_uplift: Optional[float] = None median_cannibalization: Optional[float] = None expected_uplift: Optional[float] = None # similarity-weighted, band-adjusted expected_qty_per_week: Optional[float] = None baseline_qty_per_week: Optional[float] = None baseline_source: Optional[str] = None # 'own' | 'group_avg' (new-SKU fallback) baseline_group_n: Optional[int] = None # siblings used when group_avg group_price: Optional[float] = None # parent-group avg sell price (new-SKU fallback) group_cost: Optional[float] = None # parent-group avg cost upside_uplift: Optional[float] = None # p90 scenario upside_ratio: Optional[float] = None # uplift_p90 / median upside_qty_per_week: Optional[float] = None band_mult: Optional[float] = None price_disruptor_mult: Optional[float] = None price_zone: Optional[str] = None hits: list[AnalogSkuHit] = [] note: Optional[str] = None # ---------------------------------------------------------------------- # NC30 — Croatian price-floor compliance # ---------------------------------------------------------------------- class Nc30CheckResult(BaseModel): sku: str has_data: bool # False = no NC30 ref available nc30: Optional[float] = None planned_promo_price: float delta_pct: Optional[float] = None # (planned - nc30) / nc30 × 100 ok: bool # True = compliant or skipped message: str class Nc30BatchItem(BaseModel): sku: str planned_promo_price: float # ---------------------------------------------------------------------- # Per-SKU snapshot used by the Planner # ---------------------------------------------------------------------- class SkuSnapshot(BaseModel): sku: str name: Optional[str] = None category: Optional[str] = None tier: Optional[str] = None xyz: Optional[str] = None price_retail: Optional[float] = None price_webshop: Optional[float] = None avg_sell_price: Optional[float] = None cost_price: Optional[float] = None ruc_unit: Optional[float] = None nc30: Optional[float] = None baseline_avg_weekly: float baseline_n_weeks: int on_hand: float lead_time_weeks: Optional[float] = None moq: Optional[float] = None # ---------------------------------------------------------------------- # Parent groups (product families) # ---------------------------------------------------------------------- class ParentGroupChild(BaseModel): sku: str name: Optional[str] = None tier: Optional[str] = None class ParentGroup(BaseModel): key: str display: str category: Optional[str] = None n_skus: int skus: list[ParentGroupChild] class ParentGroupsResponse(BaseModel): groups: list[ParentGroup] n_groups: int # ---------------------------------------------------------------------- # Forecaster — SKU × outcome → recommendation # ---------------------------------------------------------------------- class ForecasterSensitivity(BaseModel): discount_pct: float promo_price: Optional[float] = None predicted_uplift: float breakeven_uplift: Optional[float] = None margin_ok: bool class ForecasterHistoryRow(BaseModel): start_year: int start_week: int end_year: int end_week: int cw_label: str n_weeks: int promo_types: Optional[str] = None qty_before_avg: Optional[float] = None qty_during_avg: Optional[float] = None actual_uplift: Optional[float] = None class ForecasterResponse(BaseModel): sku: str name: Optional[str] = None category: Optional[str] = None tier: Optional[str] = None outcome: str rec_discount: Optional[float] = None rec_duration: int = 4 pred_uplift: Optional[float] = None breakeven: Optional[float] = None margin_positive: Optional[bool] = None price: Optional[float] = None promo_price: Optional[float] = None cost: Optional[float] = None baseline_avg_weekly: float = 0 promo_qty: float = 0 base_qty: float = 0 incremental_qty: float = 0 promo_revenue: float = 0 rev_delta: float = 0 promo_ruc: float = 0 ruc_delta: float = 0 n_history: int = 0 avg_uplift_hist: Optional[float] = None avg_duration_hist: Optional[float] = None sensitivity: list[ForecasterSensitivity] = [] history: list[ForecasterHistoryRow] = [] note: Optional[str] = None error: Optional[str] = None # ---------------------------------------------------------------------- # Marketing history — webshop coupon campaigns # ---------------------------------------------------------------------- class MarketingCampaignSummary(BaseModel): campaign_id: int campaign_name: str campaign_label: Optional[str] = None first_date: Optional[str] = None last_date: Optional[str] = None days_active: int n_orders: int n_coupons: int n_skus: int total_units: float total_rev_before: float total_rev_after: float total_discount: float avg_disc_pct: float class MarketingHistoryResponse(BaseModel): campaigns: list[MarketingCampaignSummary] n_campaigns: int note: Optional[str] = None class MarketingCampaignSkuRow(BaseModel): sku: str name: Optional[str] = None category: Optional[str] = None tier: Optional[str] = None n_orders: int units: float rev_before: float rev_after: float discount: float disc_pct: float class MarketingCampaignCouponRow(BaseModel): code: str first_date: Optional[str] = None last_date: Optional[str] = None days_active: int n_orders: int units: float discount: float avg_disc_pct: float class MarketingDailyRow(BaseModel): day: str code: str units: float class MarketingCampaignDetail(BaseModel): campaign_id: int skus: list[MarketingCampaignSkuRow] coupons: list[MarketingCampaignCouponRow] daily: list[MarketingDailyRow] # ---------------------------------------------------------------------- # Calendar — ports PromoCalendar/app.py dashboard # ---------------------------------------------------------------------- # Constants mirrored from PromoCalendar/promo_data.py # Source palette — hex colors matched exactly to Streamlit CALENDAR_SOURCES: dict[str, str] = { "B2C — MP (retail)": "#7C6FEE", "B2C — WEB": "#E8734A", "B2B — FMCG": "#34D399", "B2B — FITNESS": "#38BDF8", } CALENDAR_SOURCE_TO_DEPT: dict[str, str] = { "B2C — MP (retail)": "Nabava", "B2C — WEB": "Marketing", "B2B — FMCG": "Nabava", "B2B — FITNESS": "Nabava", } CALENDAR_DEPTS = ["Nabava", "Marketing"] CALENDAR_STATUSES = [ "draft", "submitted", "approved", "revision", "live", "done", "rejected", ] class CalendarCategoryCount(BaseModel): """Per-category SKU count within a campaign — drives the 'Po grupaciji' breakdown in the unified Promo Overview.""" category: str n_skus: int class PromoSkuDetail(BaseModel): """One SKU row in a promo's drill-down: code, name, regular price, promo discount %, and implied promo price.""" sku: str name: Optional[str] = None category: Optional[str] = None price: Optional[float] = None # regular price (retail ppp / avg / webshop) discount_pct: Optional[float] = None # Rabat % from the matched ERP campaign promo_price: Optional[float] = None # price × (1 − discount) class PromoSkuOverlapPromo(BaseModel): proposal_id: int proposal_name: Optional[str] = None discount_pct: Optional[float] = None class PromoSkuOverlap(BaseModel): """An SKU promoted in 2+ proposals at once (double-promo).""" sku: str name: Optional[str] = None category: Optional[str] = None promos: list[PromoSkuOverlapPromo] = [] class CalendarEntry(BaseModel): """One row from promo_proposals, shaped for the Calendar view — mirrors PROMO_COLS from PromoCalendar/promo_data.py.""" id: int name: Optional[str] = None source: Optional[str] = None type: Optional[str] = None # equiv to mechanic outcome: Optional[str] = None objective: Optional[str] = None # "logika akcije": margin|traffic|stock_clear|acquisition status: str start_year: Optional[int] = None start_week: Optional[int] = None end_year: Optional[int] = None end_week: Optional[int] = None skus: list[str] = [] # comma-separated in Streamlit; list here n_skus: int = 0 units: int = 0 category: Optional[str] = None categories: list[CalendarCategoryCount] = [] # SKU rollup by category ("po grupaciji") n_categories: int = 0 duration_weeks: Optional[int] = None # end_yw − start_yw + 1 ("trajanje") discount_min: Optional[float] = None # min Rabat % across matched ERP campaign items discount_max: Optional[float] = None # max Rabat % owner: Optional[str] = None # proposed_by display name department: Optional[str] = None # derived from source notes: Optional[str] = None log: Optional[str] = None color: Optional[str] = None # hex from source palette created_at: Optional[str] = None cw_label: Optional[str] = None start_date: Optional[str] = None # ISO date Monday of start week end_date: Optional[str] = None # ISO date Sunday of end week class CalendarConflict(BaseModel): """One overlapping (SKU × pair) conflict. Mirrors detect_conflicts() output dict from PromoCalendar/promo_data.py.""" sku: str kind: str # 'sku_overlap' | 'category_vs_sku' | 'category_vs_category' weeks: str # 'CW20–CW23' or 'CW21' year: int id_a: int id_b: int source_a: Optional[str] = None name_a: Optional[str] = None dept_a: Optional[str] = None source_b: Optional[str] = None name_b: Optional[str] = None dept_b: Optional[str] = None category_a: Optional[str] = None category_b: Optional[str] = None class CalendarStats(BaseModel): total: int pending: int # draft + submitted approved: int n_skus: int # unique SKUs across all proposals in filter n_conflicts: int class CalendarResponse(BaseModel): entries: list[CalendarEntry] n_total: int sources: dict[str, str] = {} # source name → hex color source_to_dept: dict[str, str] = {} depts: list[str] = [] note: Optional[str] = None class CalendarConflictsResponse(BaseModel): conflicts: list[CalendarConflict] n_conflicts: int class CalendarStatsResponse(BaseModel): stats: CalendarStats # ---------------------------------------------------------------------- # Approvals # ---------------------------------------------------------------------- class ApprovalRow(BaseModel): id: int proposal_id: int decision: str # 'approved' | 'rejected' | 'revision' | 'acknowledged' feedback: Optional[str] = None decided_at: Optional[str] = None reviewed_by_id: Optional[int] = None reviewed_by: Optional[str] = None class ApprovalAction(BaseModel): """Request body for POST /api/promo/calendar/approve.""" proposal_id: int decision: str # 'approve' | 'reject' | 'revision' feedback: Optional[str] = None reviewer_id: Optional[int] = None class ApprovalActionResponse(BaseModel): proposal_id: int new_status: str approval_id: int message: str class StatusChangeAction(BaseModel): status: str # any of CALENDAR_STATUSES log_line: Optional[str] = None # optional audit line reviewer_id: Optional[int] = None class StatusChangeResponse(BaseModel): proposal_id: int new_status: str message: str class AcknowledgeConflictAction(BaseModel): id_a: int id_b: int reviewer_id: Optional[int] = None