"""Forecasts from the shop's own sales history. Plain, explainable maths (no black box): sales = recent level x day-of-week pattern x a small, capped trend demand = recent units per day per product, newer weeks counting double Everything is an ESTIMATE. `reliability` says how much history stood behind it, and the screens show that. """ from collections import defaultdict from datetime import date, datetime, time, timedelta, timezone from math import ceil, sqrt from sqlalchemy import select from sqlalchemy.orm import Session from .models import Order, OrderItem, Product, PurchaseLine, PurchaseOrder, Supplier, now HISTORY_DAYS = 84 COVER_DAYS = 14 # how many days of selling a reorder should cover, on top of the supplier's lead time def _start(d: date) -> datetime: return datetime.combine(d, time.min, tzinfo=timezone.utc) def _reliability(days_of_history: int) -> str: return "too_early" if days_of_history < 14 else "early" if days_of_history < 42 else "good" def _fit(values: list[float], dates: list[date], horizon: int, today: date): """values[i] is the total for dates[i] (zero-filled, oldest first). Returns (forecast list, weekday factors).""" n = len(values) mean_all = sum(values) / n if n else 0 sums, counts = defaultdict(float), defaultdict(int) for d, v in zip(dates, values): sums[d.weekday()] += v counts[d.weekday()] += 1 weeks = n / 7 shrink = min(1.0, weeks / 4) # little history: trust the weekday pattern less factor = {} for wd in range(7): raw = (sums[wd] / counts[wd]) / mean_all if counts[wd] and mean_all else 1.0 factor[wd] = 1 + (raw - 1) * shrink adj = [v / factor[d.weekday()] if factor[d.weekday()] else v for d, v in zip(dates, values)] recent = adj[-28:] weights = list(range(1, len(recent) + 1)) # newer days count more level = sum(w * v for w, v in zip(weights, recent)) / sum(weights) if recent else 0.0 last14, prev14 = adj[-14:], adj[-28:-14] ratio = (sum(last14) / len(last14)) / (sum(prev14) / len(prev14)) if prev14 and sum(prev14) > 0 and last14 else 1.0 growth = min(1.015, max(0.985, ratio ** (1 / 14))) # a gentle daily drift, never wild resid = [v - level * factor[d.weekday()] for d, v in zip(dates[-28:], values[-28:])] sd = sqrt(sum(r * r for r in resid) / len(resid)) if resid else 0.0 out = [] for h in range(1, horizon + 1): d = today + timedelta(days=h) mid = max(0.0, level * (growth ** h) * factor[d.weekday()]) band = 1.28 * sd * sqrt(1 + h / 14) # ~80% range, widening the further out we look out.append({"date": d.isoformat(), "value": mid, "low": max(0.0, mid - band), "high": mid + band}) return out, factor def sales_forecast(db: Session, horizon: int = 14) -> dict: horizon = max(7, min(horizon, 28)) today = now().date() first = today - timedelta(days=HISTORY_DAYS - 1) rows = db.execute(select(Order.created_at, Order.total).where(Order.status != "cancelled", Order.created_at >= _start(first))).all() sales, orders = defaultdict(float), defaultdict(int) for at, total in rows: sales[at.date()] += float(total or 0) orders[at.date()] += 1 seen = sorted(sales) span = (today - seen[0]).days + 1 if seen else 0 dates = [today - timedelta(days=HISTORY_DAYS - i) for i in range(HISTORY_DAYS)] # up to yesterday: today is not finished dates = [d for d in dates if seen and d >= seen[0]] or dates[-7:] # do not count days before the shop had orders sv, ov = [sales[d] for d in dates], [float(orders[d]) for d in dates] fs, factors = _fit(sv, dates, horizon, today) fo, _ = _fit(ov, dates, horizon, today) history = [{"date": d.isoformat(), "sales": round(sales[d], 2), "orders": orders[d]} for d in (today - timedelta(days=27 - i) for i in range(28))] fc = [{"date": a["date"], "sales": round(a["value"], 2), "low": round(a["low"], 2), "high": round(a["high"], 2), "orders": round(b["value"], 1)} for a, b in zip(fs, fo)] def tot(rows_, key, n): return round(sum(r[key] for r in rows_[:n]), 2) last7 = round(sum(h["sales"] for h in history[-8:-1]), 2) last14 = round(sum(h["sales"] for h in history[-15:-1]), 2) next7, next14 = tot(fc, "sales", 7), tot(fc, "sales", 14) names = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"] best = max(factors, key=lambda k: factors[k]) quiet = min(factors, key=lambda k: factors[k]) return { "reliability": _reliability(span), "days_of_history": span, "horizon": horizon, "history": history, "forecast": fc, "next7": next7, "next14": next14 if horizon >= 14 else None, "last7": last7, "last14": last14, "change7_pct": round((next7 - last7) / last7 * 100, 1) if last7 else None, "orders_next7": round(sum(r["orders"] for r in fc[:7]), 1), "weekday_pattern": [{"day": names[k], "index": round(factors[k], 2)} for k in range(7)], "busiest_day": names[best], "quietest_day": names[quiet], } def product_demand(db: Session, horizon: int = 14) -> list[dict]: """One row per active product: how fast it sells, when it runs out, and how much to order.""" horizon = max(7, min(horizon, 56)) today = now().date() lo = _start(today - timedelta(days=28)) # whole days only: today is still being sold per_day: dict[int, dict[date, int]] = defaultdict(lambda: defaultdict(int)) for at, pid, qty in db.execute(select(Order.created_at, OrderItem.product_id, OrderItem.quantity) .join(Order, Order.id == OrderItem.order_id) .where(Order.status != "cancelled", Order.created_at >= lo)).all(): if at.date() < today: per_day[pid][at.date()] += qty on_order: dict[int, int] = defaultdict(int) for pid, qty in db.execute(select(PurchaseLine.product_id, PurchaseLine.quantity).join(PurchaseOrder) .where(PurchaseOrder.status == "ordered")).all(): on_order[pid] += qty lead = {s.id: s.lead_days for s in db.scalars(select(Supplier))} supplier = {s.id: s.name for s in db.scalars(select(Supplier))} cut = today - timedelta(days=14) out = [] for p in db.scalars(select(Product).where(Product.active).order_by(Product.name)): days = per_day.get(p.id, {}) u_new = sum(q for d, q in days.items() if d >= cut) u_old = sum(q for d, q in days.items() if d < cut) rate = (2 * u_new / 14 + u_old / 14) / 3 # units a day, newer fortnight counts double lead_days = lead.get(p.supplier_id, 7) if p.supplier_id else 7 left = p.stock / rate if rate > 0 else None need = rate * (lead_days + COVER_DAYS) - p.stock - on_order.get(p.id, 0) suggested = max(0, ceil(need)) if rate > 0 else 0 if rate == 0 and p.stock <= p.reorder_level: suggested = max(p.reorder_level * 3 - p.stock - on_order.get(p.id, 0), 1) if not on_order.get(p.id) else 0 if rate == 0: risk = "out" if p.stock == 0 and p.stock <= p.reorder_level else "no_sales" elif p.stock == 0: risk = "out" elif left < lead_days: risk = "urgent" # runs out before an order could arrive elif left < lead_days + 7: risk = "soon" else: risk = "ok" out.append({ "id": p.id, "sku": p.sku, "name": p.name, "brand": p.brand, "size": p.size, "image_url": p.image_url, "stock": p.stock, "per_day": round(rate, 2), "expected": round(rate * horizon), "days_left": round(left) if left is not None else None, "runs_out_on": (today + timedelta(days=int(left))).isoformat() if left is not None and left < 365 else None, "lead_days": lead_days, "on_order": on_order.get(p.id, 0), "suggested": suggested, "risk": risk, "supplier_id": p.supplier_id, "supplier": supplier.get(p.supplier_id, ""), "selling_days": len(days), "reliable": len(days) >= 3, }) return out