| """ |
| Predicty · Sales Forecast API (v2.2 — teaching edition) |
| ========================================================= |
| Backend สำหรับหลักสูตร Non-Degree "นักวิเคราะห์ข้อมูลและแบบจำลองเชิงพยากรณ์" |
| กรณีศึกษา: ENT Group |
| |
| ปรัชญาของเวอร์ชันนี้ — "อ่านผลอย่างซื่อสัตย์" |
| ------------------------------------------------ |
| v1 มีปัญหาเชิงการสอน 4 ข้อ ที่ v2 แก้ทั้งหมด: |
| |
| 1. confidence เดิมคือสูตรกุขึ้นเอง 1 - resid/mean → หน้าเว็บโชว์ 94% |
| ทั้งที่ข้อมูลจริง ENT พยากรณ์ผิดเฉลี่ย ~30% |
| v2: ความแม่นยำมาจาก "การทดสอบย้อนหลังจริง" (rolling-origin backtest) เท่านั้น |
| |
| 2. ช่วงความเชื่อมั่นเดิมใช้ค่าเบี่ยงเบนตัวเดียวคงที่ทุกเดือน |
| v2: ช่วงมาจาก "การกระจายของความผิดพลาดจริงที่ระยะนั้น ๆ" → กว้างขึ้นตามระยะ |
| |
| 3. guardrails เดิมบีบค่าพยากรณ์เงียบ ๆ ไว้ใน [avg/3, avg*3] |
| v2: ปิดโดยปริยาย และถ้าเปิดจะรายงานว่าเดือนไหนถูกตัดบ้าง |
| |
| 4. เลือกโมเดลจาก holdout ครั้งเดียว ≤3 เดือน แล้วไม่บอกผล |
| v2: rolling-origin หลายจุดเริ่ม + ส่ง "ตารางอันดับ" กลับไปให้ผู้เรียนเห็นทุกโมเดล |
| พร้อม skill score เทียบ baseline — ถ้าชนะ baseline ไม่ได้ ต้องกล้าบอก |
| |
| Endpoints |
| --------- |
| GET / ข้อมูลบริการ + รายการ endpoint |
| GET /health health check |
| GET /api/sample ชุดข้อมูลจริง ENT 90 เดือน (ไว้กดโหลดใน UI) |
| POST /api/forecast พยากรณ์ + ตารางเทียบโมเดล + คำเตือนคุณภาพข้อมูล |
| POST /api/backtest ทดสอบย้อนหลังอย่างเดียว (ไม่พยากรณ์อนาคต) |
| POST /api/decompose แยกองค์ประกอบ STL: trend / seasonal / residual |
| POST /api/diagnostics ADF stationarity, ACF/PACF, ตรวจคุณภาพข้อมูล |
| POST /api/describe อธิบายข้อมูลสำหรับผู้เริ่มต้น + แผนการวิเคราะห์ 6 ขั้น |
| POST /api/charts ชุดข้อมูลกราฟหลายแบบ (YoY / YTD / heatmap / growth / …) |
| POST /api/hindcast ทำนายย้อนกลับทับเส้นจริง + ตรวจความสอดคล้องของ YoY |
| POST /api/dashboard แดชบอร์ดตามบทบาท strategic / operational / analytical |
| GET /api/sample.xlsx ดาวน์โหลดข้อมูลตัวอย่างเป็น Excel |
| """ |
|
|
| import hashlib |
| import io |
| import json |
| import os |
| import time |
| import warnings |
|
|
| import numpy as np |
| import pandas as pd |
| from flask import Flask, jsonify, request |
| from flask_cors import CORS |
| from statsmodels.tsa.holtwinters import ExponentialSmoothing |
| from statsmodels.tsa.statespace.sarimax import SARIMAX |
|
|
| warnings.filterwarnings("ignore") |
|
|
| app = Flask(__name__) |
| CORS(app, resources={r"/*": {"origins": "*"}}) |
|
|
| |
| Z = {80: 1.2816, 90: 1.6449, 95: 1.9600} |
|
|
| SEASONAL_PERIOD = 12 |
| MIN_HISTORY = 6 |
| MIN_TRAIN = 36 |
| SEASONAL_MIN = 36 |
| MAX_ORIGINS = 12 |
| MIN_ORIGINS = 3 |
| BACKTEST_HORIZON = 6 |
|
|
| _CACHE = {} |
| _CACHE_MAX = 32 |
|
|
|
|
| |
| |
| |
|
|
| def to_series(rows): |
| """แปลง [{date:'2026-01', sales:123}, ...] → pandas Series รายเดือน |
| |
| คืน (series, warnings) — warnings คือสิ่งที่ผู้เรียนควรรู้ก่อนเชื่อผลพยากรณ์ |
| """ |
| warns = [] |
| df = pd.DataFrame(rows) |
|
|
| if "date" not in df.columns or "sales" not in df.columns: |
| raise ValueError("ต้องมีคอลัมน์ 'date' (YYYY-MM) และ 'sales'") |
|
|
| df["date"] = pd.to_datetime(df["date"].astype(str).str.strip() + "-01", errors="coerce") |
| df["sales"] = pd.to_numeric(df["sales"], errors="coerce") |
|
|
| bad = int(df["date"].isna().sum()) |
| if bad: |
| warns.append(f"มี {bad} แถวที่วันที่อ่านไม่ออก — ถูกตัดทิ้ง") |
| df = df.dropna(subset=["date"]) |
|
|
| dup = int(df["date"].duplicated().sum()) |
| if dup: |
| warns.append(f"มีเดือนซ้ำ {dup} รายการ — รวมกันด้วยผลบวก") |
| df = df.groupby("date", as_index=False)["sales"].sum() |
|
|
| df = df.sort_values("date").set_index("date") |
| s = df["sales"].astype(float).asfreq("MS") |
|
|
| gaps = int(s.isna().sum()) |
| if gaps: |
| warns.append( |
| f"มีเดือนที่ขาดหายไป {gaps} เดือน — เติมด้วยการประมาณค่าเชิงเส้น " |
| f"(ระวัง: ค่าที่เติมไม่ใช่ข้อมูลจริง)" |
| ) |
| s = s.interpolate().ffill().bfill() |
|
|
| return s, warns |
|
|
|
|
| def data_quality(s): |
| """ตรวจคุณภาพข้อมูล — สิ่งที่ต้องดูก่อนพยากรณ์เสมอ""" |
| v = s.values.astype(float) |
| n = len(v) |
| mean, std = float(np.mean(v)), float(np.std(v, ddof=1)) if n > 1 else 0.0 |
| cv = std / mean if mean else 0.0 |
|
|
| |
| med = float(np.median(v)) |
| mad = float(np.median(np.abs(v - med))) or 1e-9 |
| zmod = 0.6745 * (v - med) / mad |
| outliers = [ |
| {"period": s.index[i].strftime("%Y-%m"), "value": float(v[i]), "z_mad": round(float(zmod[i]), 2)} |
| for i in np.where(np.abs(zmod) > 3.5)[0] |
| ] |
|
|
| notes = [] |
| if n < MIN_TRAIN: |
| notes.append(f"ข้อมูลมี {n} เดือน — น้อยกว่า {MIN_TRAIN} เดือน (3 รอบปี) จับรูปแบบฤดูกาลได้ไม่น่าเชื่อถือ") |
| if cv > 0.35: |
| notes.append( |
| f"ข้อมูลผันผวนสูงมาก (CV = {cv:.2f}) — คาดหวังความผิดพลาดระดับหลายสิบเปอร์เซ็นต์เป็นเรื่องปกติ" |
| ) |
| if outliers: |
| notes.append(f"พบเดือนที่ค่าสุดโต่ง {len(outliers)} เดือน — ตรวจว่าเป็นดีลก้อนใหญ่จริงหรือข้อมูลผิด") |
| if n >= 4: |
| last, prev3 = float(v[-1]), float(np.mean(v[-4:-1])) |
| if prev3 > 0 and last < prev3 * 0.5: |
| notes.append( |
| f"เดือนล่าสุด ({s.index[-1].strftime('%Y-%m')}) ต่ำกว่าค่าเฉลี่ย 3 เดือนก่อนหน้ามาก " |
| f"— มักแปลว่าเดือนนั้นยังเก็บข้อมูลไม่ครบ ควรตัดออกก่อนฝึกโมเดล" |
| ) |
|
|
| return { |
| "n_months": n, |
| "start": s.index[0].strftime("%Y-%m"), |
| "end": s.index[-1].strftime("%Y-%m"), |
| "mean": round(mean, 2), |
| "median": round(med, 2), |
| "std": round(std, 2), |
| "cv": round(cv, 3), |
| "min": round(float(v.min()), 2), |
| "max": round(float(v.max()), 2), |
| "zero_months": int((v == 0).sum()), |
| "outliers": outliers, |
| "notes": notes, |
| } |
|
|
|
|
| |
| |
| |
|
|
| def fit_naive(train, h): |
| """ทายว่าเดือนหน้าเท่ากับเดือนล่าสุด — baseline ที่ง่ายที่สุด""" |
| return np.full(h, float(train.values[-1])) |
|
|
|
|
| def fit_mean3(train, h): |
| """ทายด้วยค่าเฉลี่ย 3 เดือนล่าสุด — baseline ที่ธุรกิจใช้จริงบ่อยที่สุด""" |
| return np.full(h, float(np.mean(train.values[-3:]))) |
|
|
|
|
| def fit_seasonal_naive(train, h): |
| """ทายว่าเดือนนี้ปีนี้ = เดือนเดียวกันปีที่แล้ว — baseline มาตรฐานของข้อมูลมีฤดูกาล""" |
| if len(train) < SEASONAL_PERIOD: |
| return fit_mean3(train, h) |
| last_year = train.values[-SEASONAL_PERIOD:] |
| return np.tile(last_year, int(np.ceil(h / SEASONAL_PERIOD)))[:h].astype(float) |
|
|
|
|
| def fit_linear(train, h): |
| """ลากเส้นตรงผ่านข้อมูลทั้งหมดแล้วต่อไปข้างหน้า""" |
| x = np.arange(len(train)) |
| coef = np.polyfit(x, train.values, 1) |
| return np.polyval(coef, np.arange(len(train), len(train) + h)) |
|
|
|
|
| def fit_holtwinters(train, h): |
| """Exponential Smoothing: ระดับ + แนวโน้ม + ฤดูกาล (ถ้าข้อมูลยาวพอ)""" |
| use_seasonal = len(train) >= SEASONAL_MIN |
| m = ExponentialSmoothing( |
| train, |
| trend="add", |
| seasonal="add" if use_seasonal else None, |
| seasonal_periods=SEASONAL_PERIOD if use_seasonal else None, |
| initialization_method="estimated", |
| ).fit(optimized=True) |
| return np.asarray(m.forecast(h), dtype=float) |
|
|
|
|
| def fit_sarima(train, h): |
| """SARIMA(1,1,1)(1,1,1)[12] — โมเดลอนุกรมเวลามาตรฐาน""" |
| seasonal_order = (1, 1, 1, SEASONAL_PERIOD) if len(train) >= SEASONAL_MIN else (0, 0, 0, 0) |
| m = SARIMAX( |
| train, |
| order=(1, 1, 1), |
| seasonal_order=seasonal_order, |
| enforce_stationarity=False, |
| enforce_invertibility=False, |
| ).fit(disp=False) |
| return np.asarray(m.get_forecast(h).predicted_mean, dtype=float) |
|
|
|
|
| def _lag_frame(values, lags=(1, 2, 3, 12), months=None): |
| """แปลงอนุกรมเวลาเป็นตารางแบบ supervised learning |
| |
| บทเรียน: ML ทั่วไปไม่รู้จัก "เวลา" เราต้องป้อนอดีตให้มันเองผ่าน lag features |
| """ |
| import numpy as _np |
| v = _np.asarray(values, dtype=float) |
| n, mx = len(v), max(lags) |
| rows, ys = [], [] |
| for t in range(mx, n): |
| row = [v[t - L] for L in lags] |
| if months is not None: |
| row.append(months[t]) |
| rows.append(row) |
| ys.append(v[t]) |
| return _np.array(rows, dtype=float), _np.array(ys, dtype=float) |
|
|
|
|
| def fit_ml_rf(train, h): |
| """Random Forest บน lag features — ตัวแทนฝั่ง Machine Learning |
| |
| พยากรณ์หลายงวดแบบ recursive: ทำนายเดือนถัดไป → เอาค่าที่ทำนายได้ไปเป็น input |
| ของเดือนถัดไปอีกที ความผิดพลาดจึงสะสมตามระยะ ซึ่งเป็นข้อจำกัดจริงของวิธีนี้ |
| """ |
| from sklearn.ensemble import RandomForestRegressor |
|
|
| LAGS = (1, 2, 3, 12) |
| v = np.asarray(train.values, dtype=float) |
| if len(v) < max(LAGS) + 8: |
| return fit_mean3(train, h) |
|
|
| months = list(pd.DatetimeIndex(train.index).month) |
| X, y = _lag_frame(v, LAGS, months) |
| model = RandomForestRegressor(n_estimators=200, max_depth=8, min_samples_leaf=3, |
| random_state=7, n_jobs=1).fit(X, y) |
|
|
| hist = list(v) |
| last_month = months[-1] |
| out = [] |
| for _ in range(h): |
| last_month = last_month % 12 + 1 |
| row = [hist[-L] for L in LAGS] + [last_month] |
| pred = float(model.predict(np.array([row], dtype=float))[0]) |
| out.append(pred) |
| hist.append(pred) |
| return np.asarray(out, dtype=float) |
|
|
|
|
| MODELS = { |
| "naive": fit_naive, |
| "mean3": fit_mean3, |
| "seasonal_naive": fit_seasonal_naive, |
| "linear": fit_linear, |
| "holtwinters": fit_holtwinters, |
| "sarima": fit_sarima, |
| "ml_rf": fit_ml_rf, |
| } |
|
|
| |
| BASELINES = ["naive", "mean3", "seasonal_naive"] |
| BENCHMARK = "seasonal_naive" |
|
|
| MODEL_INFO = { |
| "naive": {"th": "ทายเท่าเดือนล่าสุด", "kind": "baseline", |
| "how": "ค่าพยากรณ์ทุกเดือน = ค่าของเดือนสุดท้ายในข้อมูล"}, |
| "mean3": {"th": "ค่าเฉลี่ย 3 เดือนล่าสุด", "kind": "baseline", |
| "how": "ค่าพยากรณ์ทุกเดือน = ค่าเฉลี่ยของ 3 เดือนสุดท้าย"}, |
| "seasonal_naive": {"th": "ทายเท่าเดือนเดียวกันปีที่แล้ว", "kind": "baseline", |
| "how": "ม.ค.2570 = ม.ค.2569 — จับฤดูกาลได้โดยไม่ต้องมีโมเดล"}, |
| "linear": {"th": "เส้นแนวโน้มตรง", "kind": "model", |
| "how": "ลากเส้นตรงที่พอดีกับข้อมูลที่สุด แล้วต่อไปข้างหน้า (ไม่มีฤดูกาล)"}, |
| "holtwinters": {"th": "Holt-Winters (Exponential Smoothing)", "kind": "model", |
| "how": "ให้น้ำหนักข้อมูลใหม่มากกว่าเก่า แยกเป็น ระดับ + แนวโน้ม + ฤดูกาล"}, |
| "sarima": {"th": "SARIMA (1,1,1)(1,1,1)[12]", "kind": "model", |
| "how": "ใช้ค่าในอดีตและความผิดพลาดในอดีตทำนาย พร้อมส่วนฤดูกาล 12 เดือน"}, |
| "ml_rf": {"th": "Random Forest (Machine Learning)", "kind": "model", |
| "how": "ป้อนค่าย้อนหลัง 1/2/3/12 เดือน + เดือนของปี ให้ต้นไม้ตัดสินใจหลายร้อยต้นโหวตกัน"}, |
| } |
|
|
| |
| ALIASES = {"arima": "sarima", "hw": "holtwinters", "ets": "holtwinters"} |
|
|
|
|
| |
| |
| |
|
|
| def _safe_ape(actual, pred): |
| a = np.asarray(actual, dtype=float) |
| p = np.asarray(pred, dtype=float) |
| denom = np.where(np.abs(a) < 1e-9, np.nan, a) |
| return np.abs((a - p) / denom) * 100 |
|
|
|
|
| def score(actual, pred): |
| """MAPE / MdAPE / MAE / RMSE |
| |
| MAPE = ผิดเฉลี่ยกี่ % — สื่อสารกับผู้บริหารง่ายที่สุด แต่ถูกเดือนที่ยอดต่ำดึงให้พองได้ |
| MdAPE = ค่ากลางของ % ที่ผิด — ทนต่อเดือนสุดโต่งกว่า |
| MAE = ผิดเฉลี่ยกี่บาท — ใช้ตัดสินใจธุรกิจได้ตรงกว่า % |
| RMSE = ลงโทษความผิดพลาดก้อนใหญ่หนักกว่า MAE |
| """ |
| a = np.asarray(actual, dtype=float) |
| p = np.asarray(pred, dtype=float) |
| ape = _safe_ape(a, p) |
| return { |
| "mape": float(np.nanmean(ape)), |
| "mdape": float(np.nanmedian(ape)), |
| "mae": float(np.mean(np.abs(a - p))), |
| "rmse": float(np.sqrt(np.mean((a - p) ** 2))), |
| } |
|
|
|
|
| def rolling_origin_backtest(s, horizon=BACKTEST_HORIZON, max_origins=MAX_ORIGINS, models=None, |
| train_window=None, log_transform=False): |
| """ทดสอบย้อนหลังแบบเลื่อนจุดเริ่ม — วิธีวัดความแม่นยำที่ถูกต้องของอนุกรมเวลา |
| |
| หลักการ: ทำเสมือนย้อนเวลากลับไปยืนที่จุดหนึ่งในอดีต แล้วพยากรณ์ไปข้างหน้า |
| โดยเห็นเฉพาะข้อมูลก่อนจุดนั้น จากนั้นเทียบกับสิ่งที่เกิดขึ้นจริง — ทำซ้ำหลายจุด |
| |
| ทำไมต้องเลื่อนหลายจุด: วัดครั้งเดียวคือการวัด "ดวง" ไม่ใช่ "ฝีมือ" |
| |
| คืน: {model: {"metrics": {...}, "errors_by_h": {h: [error, ...]}}} |
| """ |
| models = models or list(MODELS.keys()) |
| n = len(s) |
| FLOOR = max(MIN_HISTORY, SEASONAL_PERIOD + 3) |
| min_train = max(MIN_TRAIN, SEASONAL_PERIOD + 3) |
|
|
| |
| want = n - horizon + 1 - MIN_ORIGINS |
| shortened = False |
| if want < min_train: |
| new_min = max(FLOOR, want) |
| shortened = new_min < min_train |
| min_train = new_min |
|
|
| if n < min_train + horizon: |
| return None, "ข้อมูลสั้นเกินไปสำหรับการทดสอบย้อนหลัง" |
|
|
| all_origins = list(range(min_train, n - horizon + 1)) |
| if len(all_origins) < MIN_ORIGINS: |
| |
| return None, (f"มีจุดทดสอบย้อนหลังได้เพียง {len(all_origins)} จุด " |
| f"(ต้องการอย่างน้อย {MIN_ORIGINS}) — วัดความแม่นยำอย่างน่าเชื่อถือไม่ได้") |
| if len(all_origins) > max_origins: |
| idx = np.linspace(0, len(all_origins) - 1, max_origins).round().astype(int) |
| origins = [all_origins[i] for i in sorted(set(idx))] |
| else: |
| origins = all_origins |
|
|
| out = {m: {"errors_by_h": {h: [] for h in range(1, horizon + 1)}, |
| "actual": [], "pred": [], |
| "agg_actual": [], "agg_pred": [], |
| "failed": 0} for m in models} |
|
|
| for o in origins: |
| train, test = s.iloc[:o], s.iloc[o:o + horizon] |
| if train_window and len(train) > train_window: |
| train = train.iloc[-train_window:] |
| fit_on = np.log(train) if log_transform else train |
| for name in models: |
| try: |
| fc = MODELS[name](fit_on, horizon) |
| if log_transform: |
| fc = np.exp(fc) |
| if not np.all(np.isfinite(fc)): |
| raise ValueError("ผลพยากรณ์ไม่ใช่ตัวเลขที่ใช้ได้") |
| except Exception: |
| out[name]["failed"] += 1 |
| continue |
| out[name]["actual"].extend(test.values.tolist()) |
| out[name]["pred"].extend(fc.tolist()) |
| |
| out[name]["agg_actual"].append(float(np.sum(test.values))) |
| out[name]["agg_pred"].append(float(np.sum(fc))) |
| for h in range(1, horizon + 1): |
| out[name]["errors_by_h"][h].append(float(test.values[h - 1] - fc[h - 1])) |
|
|
| result = {} |
| for name, d in out.items(): |
| if len(d["actual"]) < horizon: |
| continue |
| result[name] = { |
| "metrics": score(d["actual"], d["pred"]), |
| "metrics_aggregate": score(d["agg_actual"], d["agg_pred"]), |
| "errors_by_h": d["errors_by_h"], |
| "n_origins": len(origins), |
| "failed_fits": d["failed"], |
| } |
|
|
| meta = { |
| "n_origins": len(origins), |
| "horizon": horizon, |
| "min_train_months": min_train, |
| "train_window": train_window or "expanding (ใช้ข้อมูลทั้งหมดตั้งแต่ต้น)", |
| "log_transform": bool(log_transform), |
| "train_window_shortened": shortened, |
| "origin_periods": [s.index[o - 1].strftime("%Y-%m") for o in origins], |
| } |
| if shortened: |
| meta["note"] = ( |
| f"หน้าต่างเทรนถูกย่อเหลือ {min_train} เดือน (ต่ำกว่า {MIN_TRAIN} เดือน = 3 รอบปี) " |
| "เพื่อให้มีจุดทดสอบพอ — ตัวเลขความแม่นยำจึงมีความไม่แน่นอนสูง" |
| ) |
| return result, meta |
|
|
|
|
| def build_leaderboard(bt): |
| """เรียงอันดับโมเดลตาม MAPE + คำนวณ skill score เทียบ baseline |
| |
| skill = 1 - MAPE(model) / MAPE(seasonal_naive) |
| > 0 ดีกว่า baseline |
| = 0 เท่า baseline |
| < 0 แย่กว่า baseline (โมเดลไม่มีประโยชน์ — ต้องกล้าบอก) |
| """ |
| if not bt: |
| return [], None |
|
|
| bench = bt.get(BENCHMARK) or bt.get("mean3") |
| bench_mape = bench["metrics"]["mape"] if bench else None |
|
|
| rows = [] |
| for name, d in bt.items(): |
| m = d["metrics"] |
| skill = None |
| if bench_mape and bench_mape > 0: |
| skill = 1 - m["mape"] / bench_mape |
| rows.append({ |
| "model": name, |
| "label": MODEL_INFO.get(name, {}).get("th", name), |
| "kind": MODEL_INFO.get(name, {}).get("kind", "model"), |
| "how": MODEL_INFO.get(name, {}).get("how", ""), |
| "mape": round(m["mape"], 2), |
| "mdape": round(m["mdape"], 2), |
| "mae": round(m["mae"], 2), |
| "rmse": round(m["rmse"], 2), |
| "mape_total": round(d["metrics_aggregate"]["mape"], 2), |
| "skill_vs_baseline": None if skill is None else round(skill, 3), |
| "beats_baseline": None if skill is None else bool(skill > 0), |
| }) |
| rows.sort(key=lambda r: r["mape"]) |
| return rows, bench_mape |
|
|
|
|
| def reliability_label(mape, best_model, leaderboard): |
| """แปลตัวเลข MAPE เป็นภาษาที่ผู้บริหารและ นศ. ใช้ตัดสินใจได้จริง""" |
| beat = any(r["kind"] == "model" and r["beats_baseline"] for r in leaderboard) |
| winner_is_baseline = MODEL_INFO.get(best_model, {}).get("kind") == "baseline" |
|
|
| if mape < 10: |
| level, th = "high", "แม่นสูง — ใช้วางแผนรายเดือนได้" |
| elif mape < 20: |
| level, th = "medium", "พอใช้ — ใช้ดูทิศทาง ไม่ควรผูกกับตัวเลขเป๊ะ ๆ" |
| elif mape < 35: |
| level, th = "low", "ต่ำ — ใช้ได้แค่ดูแนวโน้มกว้าง ๆ ห้ามใช้ตั้งเป้ารายเดือน" |
| else: |
| level, th = "very_low", "ต่ำมาก — ข้อมูลผันผวนเกินกว่าจะพยากรณ์รายเดือนได้" |
|
|
| honest = [] |
| if winner_is_baseline: |
| honest.append( |
| "โมเดลที่ชนะคือ baseline (วิธีง่าย ๆ ที่ไม่ต้องใช้ AI) " |
| "— แปลว่าข้อมูลชุดนี้ยังไม่มีสัญญาณพอให้โมเดลซับซ้อนได้เปรียบ" |
| ) |
| if not beat: |
| honest.append( |
| "ไม่มีโมเดลใดเอาชนะ baseline ได้ — ผลลัพธ์ที่ซื่อสัตย์คือ " |
| "\"ยังพยากรณ์รายเดือนได้ไม่ดีกว่าการเดาแบบง่าย\"" |
| ) |
| if mape >= 25: |
| honest.append( |
| f"ค่าพยากรณ์ผิดเฉลี่ย {mape:.0f}% — ควรรายงานเป็น 'ช่วง' ไม่ใช่ 'ตัวเลขเดียว' " |
| "และควรรวมยอดเป็นรายไตรมาสซึ่งแม่นกว่ารายเดือน" |
| ) |
| return {"level": level, "th": th, "honest_notes": honest, "any_model_beats_baseline": beat} |
|
|
|
|
| |
| |
| |
|
|
| def empirical_intervals(mean, errors_by_h, level, fallback_std): |
| """สร้างช่วงความเชื่อมั่นจาก "ความผิดพลาดที่เคยเกิดขึ้นจริง" ที่ระยะนั้น ๆ |
| |
| นี่คือหัวใจเชิงการสอน: ช่วงกว้างขึ้นตามระยะ "เพราะวัดได้จริงว่ามันกว้างขึ้น" |
| ไม่ใช่เพราะสูตรบอกให้กว้าง |
| |
| วิธีทำ |
| 1. จาก backtest เก็บความผิดพลาดที่ระยะ h = 1, 2, 3, ... แยกกอง |
| 2. σ(h) = ส่วนเบี่ยงเบนมาตรฐานของกองนั้น |
| 3. บังคับให้ σ ไม่ลดลงตามระยะ (running max) — เพราะจุดทดสอบมีจำกัด |
| σ ดิบจึงแกว่ง ถ้าไม่บังคับ จะได้ช่วงเดือนที่ 5 แคบกว่าเดือนที่ 2 ซึ่งไม่สมเหตุผล |
| 4. เลยระยะที่ทดสอบไว้ → ขยายต่อด้วย √(h/h_max) ตามทฤษฎี random walk |
| |
| หมายเหตุที่ควรบอกผู้เรียน: ใช้ σ (สมมาตร) แทน quantile ดิบ เพราะจุดทดสอบระดับ |
| ~10 จุดยังน้อยเกินกว่าจะประมาณ quantile หางได้เสถียร — บอกความไม่แน่นอน |
| ของ "การประมาณความไม่แน่นอน" ตรง ๆ ดีกว่าแกล้งทำเป็นแม่น |
| """ |
| z = Z.get(int(level), Z[80]) |
| max_h = max(errors_by_h.keys()) if errors_by_h else 0 |
|
|
| |
| sigma, sigma_raw, n_used, method = {}, {}, {}, "normal_approx" |
| running = 0.0 |
| for h in range(1, max_h + 1): |
| e = np.asarray(errors_by_h.get(h, []), dtype=float) |
| e = e[np.isfinite(e)] |
| n_used[h] = int(len(e)) |
| if len(e) >= 5: |
| raw = float(np.std(e, ddof=1)) |
| sigma_raw[h] = round(raw, 2) |
| running = max(running, raw) |
| sigma[h] = running |
| method = "empirical_sigma_by_horizon" |
|
|
| lower, upper, sigma_out = [], [], [] |
| for i, m in enumerate(mean, start=1): |
| if sigma: |
| h = min(i, max(sigma.keys())) |
| sd = sigma[h] |
| if i > max(sigma.keys()): |
| sd = sd * np.sqrt(i / max(sigma.keys())) |
| else: |
| sd = fallback_std * np.sqrt(i) |
| spread = z * sd |
| lower.append(max(0.0, float(m - spread))) |
| upper.append(float(m + spread)) |
| sigma_out.append(round(float(sd), 2)) |
|
|
| detail = { |
| "method": method, |
| "z": z, |
| "sigma_by_horizon": sigma_out, |
| "sigma_raw_by_horizon": sigma_raw, |
| "origins_per_horizon": n_used, |
| "note": ( |
| "σ ดิบวัดจากความผิดพลาดจริง แล้วบังคับไม่ให้ลดลงตามระยะ — " |
| "ถ้า σ ดิบของเดือนที่ไกลกว่ากลับเล็กลง แปลว่าจุดทดสอบยังน้อย ไม่ใช่ว่าพยากรณ์ไกลแล้วแม่นขึ้น" |
| ), |
| } |
| return np.array(lower), np.array(upper), detail |
|
|
|
|
| def apply_guardrails(fc, history): |
| """บีบค่าพยากรณ์ให้อยู่ในกรอบที่สมเหตุสมผล — ปิดโดยปริยาย |
| |
| ข้อควรระวังเชิงการสอน: การบีบค่าแบบนี้ทำให้กราฟ "ดูดี" ขึ้น |
| แต่ซ่อนความจริงว่าโมเดลกำลังให้คำตอบที่เพี้ยน — ถ้าเปิดใช้ต้องรายงานเสมอ |
| """ |
| avg = float(np.mean(history)) |
| if avg <= 0: |
| return fc, [] |
| lo, hi = avg / 3, avg * 3 |
| clipped = [i for i, v in enumerate(fc) if v < lo or v > hi] |
| return np.clip(fc, lo, hi), clipped |
|
|
|
|
| def next_periods(last_date, n): |
| return [(last_date + pd.DateOffset(months=i + 1)).strftime("%Y-%m") for i in range(n)] |
|
|
|
|
| def _hash(rows, extra=""): |
| return hashlib.md5((json.dumps(rows, sort_keys=True, default=str) + extra).encode()).hexdigest() |
|
|
|
|
| def _cached_backtest(rows, s, horizon): |
| key = _hash(rows, f"|bt{horizon}") |
| if key in _CACHE: |
| return _CACHE[key] |
| t0 = time.time() |
| bt, meta = rolling_origin_backtest(s, horizon=horizon) |
| if isinstance(meta, dict): |
| meta["elapsed_sec"] = round(time.time() - t0, 2) |
| if len(_CACHE) >= _CACHE_MAX: |
| _CACHE.pop(next(iter(_CACHE))) |
| _CACHE[key] = (bt, meta) |
| return bt, meta |
|
|
|
|
| def explain_forecast(s, used, mean, labels, leaderboard, acc, ci_detail, quality, level): |
| """อธิบายว่าทำไมผลพยากรณ์ถึงออกมาแบบนี้ — เป็นภาษาที่คนไม่ได้เรียนสถิติอ่านรู้เรื่อง |
| |
| ตัวเลขพยากรณ์เปล่า ๆ ไม่ช่วยให้ใครตัดสินใจ สิ่งที่ช่วยคือการรู้ว่า |
| ค่านี้มาจากไหน · เชื่อได้แค่ไหน · และมันทำอะไรไม่ได้บ้าง |
| """ |
| v = np.asarray(s.values, dtype=float) |
| reasons = [] |
|
|
| |
| if leaderboard: |
| me = next((r for r in leaderboard if r["model"] == used), None) |
| bench = next((r for r in leaderboard if r["model"] == BENCHMARK), None) |
| if me: |
| t = (f"จากการทดสอบย้อนหลัง โมเดล \"{me['label']}\" ทำผิดพลาดเฉลี่ย {me['mape']}% " |
| f"ต่ำที่สุดในบรรดา {len(leaderboard)} วิธีที่ลอง") |
| if bench and me["skill_vs_baseline"] is not None: |
| if me["skill_vs_baseline"] > 0: |
| t += (f" และดีกว่าการเดาแบบง่าย ({bench['label']} {bench['mape']}%) " |
| f"อยู่ {me['skill_vs_baseline'] * 100:.0f}%") |
| else: |
| t += (f" แต่ยัง \"ไม่ดีกว่า\" การเดาแบบง่าย ({bench['label']} {bench['mape']}%) " |
| "— แปลว่าข้อมูลชุดนี้ยังไม่มีสัญญาณพอให้โมเดลได้เปรียบ") |
| reasons.append({"title": "ทำไมถึงเลือกโมเดลนี้", "text": t}) |
|
|
| |
| if used == "naive": |
| why = f"ทุกเดือนถูกตั้งให้เท่ากับเดือนล่าสุด ({v[-1]:,.0f} บาท) เส้นพยากรณ์จึงแบนราบ" |
| elif used == "mean3": |
| why = (f"ทุกเดือนถูกตั้งให้เท่ากับค่าเฉลี่ย 3 เดือนล่าสุด " |
| f"({v[-3]:,.0f} · {v[-2]:,.0f} · {v[-1]:,.0f} → {np.mean(v[-3:]):,.0f} บาท) " |
| "เส้นจึงแบนราบ") |
| elif used == "seasonal_naive": |
| ref = [(s.index[-SEASONAL_PERIOD + i].strftime("%b %Y"), v[-SEASONAL_PERIOD + i]) |
| for i in range(min(len(mean), SEASONAL_PERIOD))] |
| why = ("ค่าพยากรณ์ถูกคัดลอกมาจากเดือนเดียวกันของปีที่แล้วโดยตรง: " |
| + " · ".join(f"{d} = {x:,.0f}" for d, x in ref[:4]) |
| + (" …" if len(ref) > 4 else "") |
| + " — รูปร่างกราฟจึงเหมือนปีที่แล้วเป๊ะ ๆ รวมทั้งเดือนที่สูงผิดปกติด้วย") |
| elif used == "linear": |
| x = np.arange(len(v)) |
| slope, intercept = np.polyfit(x, v, 1) |
| dirn = "เพิ่มขึ้น" if slope > 0 else "ลดลง" |
| why = (f"ลากเส้นตรงที่พอดีกับข้อมูล {len(v)} เดือนที่สุด แล้วต่อออกไป " |
| f"— เส้นนี้{dirn}เดือนละ {abs(slope):,.0f} บาท " |
| f"(ประมาณ {abs(slope) * 12 / np.mean(v) * 100:.1f}% ต่อปี) " |
| "ค่าพยากรณ์จึงค่อย ๆ ไต่ไปทางเดียว ไม่มีการขึ้นลงรายเดือน") |
| elif used == "holtwinters": |
| why = ("ให้น้ำหนักข้อมูลใหม่มากกว่าข้อมูลเก่าแบบลดหลั่น แล้วแยกจำ 3 อย่าง " |
| "คือระดับ แนวโน้ม และรูปแบบฤดูกาล — ค่าที่ได้จึงเป็นระดับล่าสุด " |
| "บวกแนวโน้ม บวกส่วนต่างประจำเดือนนั้น") |
| elif used == "sarima": |
| why = ("ใช้ทั้งค่าในอดีตและความผิดพลาดในอดีตของตัวเองมาทำนายเดือนถัดไป " |
| "พร้อมส่วนฤดูกาล 12 เดือน") |
| elif used == "ml_rf": |
| why = ("ป้อนยอดย้อนหลัง 1 · 2 · 3 · 12 เดือน และเลขเดือนของปี ให้ต้นไม้ตัดสินใจ 200 ต้นโหวตกัน " |
| "เดือนถัดไปทำนายจากค่าจริง ส่วนเดือนที่ไกลออกไปทำนายจากค่าที่ตัวเองเพิ่งทายไว้ " |
| "— ความผิดพลาดจึงสะสมตามระยะ") |
| else: |
| why = "ค่าพยากรณ์คำนวณจากรูปแบบที่โมเดลจับได้จากข้อมูลในอดีต" |
| reasons.append({"title": "ตัวเลขที่ได้มาจากอะไร", "text": why}) |
|
|
| |
| last3, fc3 = float(np.mean(v[-3:])), float(np.mean(mean[:min(3, len(mean))])) |
| ch3 = (fc3 - last3) / last3 * 100 if last3 else 0 |
| t = (f"ค่าเฉลี่ย 3 เดือนแรกที่พยากรณ์ ({fc3:,.0f} บาท) " |
| f"{'สูงกว่า' if ch3 >= 0 else 'ต่ำกว่า'} 3 เดือนล่าสุดที่เกิดขึ้นจริง ({last3:,.0f} บาท) " |
| f"อยู่ {abs(ch3):.1f}%") |
| if len(v) >= SEASONAL_PERIOD + 3: |
| ly = float(np.mean(v[-SEASONAL_PERIOD - 3:-SEASONAL_PERIOD])) |
| if ly: |
| t += f" · เทียบช่วงเดียวกันของปีก่อน ({ly:,.0f} บาท) ต่างกัน {(fc3 - ly) / ly * 100:+.1f}%" |
| reasons.append({"title": "ทิศทางเทียบกับของจริง", "text": t}) |
|
|
| |
| width = float(np.mean(np.asarray(ci_detail["sigma_by_horizon"], dtype=float))) * 2 * ci_detail["z"] |
| pct = width / float(np.mean(mean)) * 100 if np.mean(mean) else 0 |
| t = (f"ช่วง {level}% กว้างเฉลี่ย {width:,.0f} บาท (ประมาณ {pct:.0f}% ของค่าพยากรณ์) " |
| f"เพราะยอดรายเดือนจริงแกว่งแรงมาก — ค่าความผันผวน CV = {quality['cv']} " |
| f"และเดือนต่ำสุดกับสูงสุดต่างกัน {quality['max'] / max(quality['min'], 1):.0f} เท่า") |
| if ci_detail.get("method") == "empirical_sigma_by_horizon": |
| t += " ความกว้างนี้วัดจากความผิดพลาดที่เกิดขึ้นจริงในการทดสอบย้อนหลัง ไม่ได้มาจากสูตร" |
| reasons.append({"title": "ทำไมช่วงถึงกว้างขนาดนี้", "text": t}) |
|
|
| |
| cant = [] |
| if acc and acc["mape"] >= 20: |
| cant.append(f"พยากรณ์รายเดือนผิดเฉลี่ย {acc['mape']:.0f}% จึงใช้ตั้งเป้ารายเดือนไม่ได้") |
| if quality.get("outliers"): |
| months = ", ".join(o["period"] for o in quality["outliers"][:3]) |
| cant.append(f"ทำนายดีลก้อนใหญ่แบบที่เกิดใน {months} ไม่ได้ เพราะดีลเหล่านั้นไม่ได้เป็นรูปแบบซ้ำ") |
| cant.append("ไม่รู้เรื่องราคาคู่แข่ง งบลูกค้า หรือ pipeline ที่กำลังจะปิด " |
| "— ปัจจัยเหล่านี้ไม่ได้อยู่ในข้อมูลที่ป้อนเข้ามา") |
| reasons.append({"title": "สิ่งที่ผลนี้ทำไม่ได้", "text": " · ".join(cant)}) |
|
|
| |
| lvl = acc["reliability"]["level"] if acc and acc.get("reliability") else "unknown" |
| headline = { |
| "high": "ผลนี้เชื่อถือได้ในระดับที่ใช้วางแผนรายเดือนได้", |
| "medium": "ผลนี้ใช้ดูทิศทางได้ แต่อย่าผูกกับตัวเลขเป๊ะ ๆ", |
| "low": "ผลนี้ใช้ได้แค่ดูแนวโน้มกว้าง ๆ ให้ตั้งเป้าเป็นยอดรวมไตรมาสแทนรายเดือน", |
| "very_low": "ข้อมูลผันผวนเกินกว่าจะพยากรณ์รายเดือนได้ ให้ใช้เป็นช่วงเท่านั้น", |
| }.get(lvl, "ยังประเมินความน่าเชื่อถือของผลนี้ไม่ได้ เพราะข้อมูลสั้นเกินไป") |
|
|
| return {"headline": headline, "reasons": reasons} |
|
|
|
|
| |
| |
| |
|
|
| @app.route("/") |
| def root(): |
| return jsonify({ |
| "status": "ok", |
| "service": "predicty-forecast", |
| "version": "2.2-teaching", |
| "principle": "ความแม่นยำทุกตัวเลขในนี้มาจากการทดสอบย้อนหลังจริง ไม่มีค่าที่คำนวณขึ้นเองให้ดูดี", |
| "endpoints": { |
| "GET /health": "health check", |
| "GET /api/sample": "ชุดข้อมูลจริง ENT 90 เดือน", |
| "POST /api/forecast": "พยากรณ์ + ตารางเทียบโมเดล + คำเตือน", |
| "POST /api/backtest": "ทดสอบย้อนหลังอย่างเดียว", |
| "POST /api/decompose": "แยก trend / seasonal / residual (STL)", |
| "POST /api/diagnostics": "ADF, ACF/PACF, คุณภาพข้อมูล", |
| "POST /api/describe": "อธิบายข้อมูล: มาจากไหน ถึงไหน คอลัมน์แปลว่าอะไร พร้อมแผนวิเคราะห์ทีละขั้น", |
| "POST /api/charts": "ชุดข้อมูลกราฟหลายแบบ: YoY, YTD, heatmap, growth, histogram, box plot", |
| "POST /api/hindcast": "ทำนายย้อนกลับทับเส้นจริง + ตรวจความสอดคล้องของ YoY", |
| "POST /api/dashboard": "แดชบอร์ดตามบทบาท: strategic / operational / analytical — คนละตัวเลข คนละข้อเสนอ", |
| "GET /api/sample.xlsx": "ดาวน์โหลดข้อมูลตัวอย่าง 90 เดือนเป็นไฟล์ Excel พร้อมพจนานุกรมคอลัมน์", |
| }, |
| "roles": {k: v["th"] for k, v in ROLES.items()}, |
| "models": MODEL_INFO, |
| }) |
|
|
|
|
| @app.route("/health") |
| def health(): |
| return jsonify({"status": "ok", "version": "2.2-teaching"}) |
|
|
|
|
| @app.route("/api/sample") |
| def sample(): |
| """ข้อมูลจริง ENT — ให้ UI กดโหลดได้ในคลิกเดียว""" |
| path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "sample_ent.json") |
| try: |
| with open(path, encoding="utf-8") as f: |
| return jsonify(json.load(f)) |
| except FileNotFoundError: |
| return jsonify({"error": "ไม่พบไฟล์ sample_ent.json ใน Space"}), 404 |
|
|
|
|
| @app.route("/api/forecast", methods=["POST", "OPTIONS"]) |
| def forecast(): |
| if request.method == "OPTIONS": |
| return ("", 204) |
| try: |
| body = request.get_json(force=True) or {} |
| rows = body.get("data", []) |
| periods = max(1, min(24, int(body.get("periods", 6)))) |
| want = str(body.get("model", "auto")).lower() |
| want = ALIASES.get(want, want) |
| level = int(body.get("interval", 80)) |
| level = level if level in Z else 80 |
| use_guardrails = bool(body.get("guardrails", False)) |
| drop_last = bool(body.get("drop_last_incomplete", False)) |
|
|
| if len(rows) < MIN_HISTORY: |
| return jsonify({"error": f"ต้องการข้อมูลย้อนหลังอย่างน้อย {MIN_HISTORY} เดือน"}), 400 |
|
|
| s, warns = to_series(rows) |
| if drop_last and len(s) > MIN_HISTORY: |
| warns.append(f"ตัดเดือนสุดท้าย ({s.index[-1].strftime('%Y-%m')}) ออกตามที่ร้องขอ") |
| s = s.iloc[:-1] |
| if len(s) < MIN_HISTORY: |
| return jsonify({"error": f"หลังทำความสะอาดแล้วเหลือ {len(s)} เดือน — ไม่พอ"}), 400 |
|
|
| quality = data_quality(s) |
|
|
| |
| bt, bt_meta = _cached_backtest(rows if not drop_last else rows[:-1], |
| s, min(BACKTEST_HORIZON, max(1, periods))) |
| leaderboard, bench_mape = build_leaderboard(bt) if bt else ([], None) |
| if bt and isinstance(bt_meta, dict) and bt_meta.get("train_window_shortened"): |
| warns.append(bt_meta["note"]) |
| if not bt: |
| why = bt_meta if isinstance(bt_meta, str) else "ข้อมูลสั้นเกินไป" |
| warns.append( |
| f"{why} — ระบบจึงบอก 'ความแม่นยำ' ไม่ได้เลย " |
| "ให้ถือว่าผลพยากรณ์นี้ยังพิสูจน์ไม่ได้ และห้ามนำไปตั้งเป้า" |
| ) |
|
|
| |
| if want in MODELS: |
| used, selection = want, "ผู้ใช้เลือกเอง" |
| elif leaderboard: |
| used, selection = leaderboard[0]["model"], "เลือกอัตโนมัติจาก MAPE ต่ำสุดในการทดสอบย้อนหลัง" |
| else: |
| used = "holtwinters" if len(s) >= SEASONAL_MIN else "linear" |
| selection = "ข้อมูลสั้นเกินกว่าจะทดสอบย้อนหลัง — เลือกตามความยาวข้อมูล" |
|
|
| |
| fallback_used = None |
| try: |
| mean = MODELS[used](s, periods) |
| if not np.all(np.isfinite(mean)): |
| raise ValueError("ผลพยากรณ์ไม่ใช่ตัวเลขที่ใช้ได้") |
| except Exception as e: |
| fallback_used = f"{used} ล้มเหลว ({type(e).__name__}) — ถอยไปใช้ seasonal_naive" |
| used = "seasonal_naive" |
| mean = MODELS[used](s, periods) |
|
|
| |
| errors_by_h = bt.get(used, {}).get("errors_by_h", {}) if bt else {} |
| resid_std = float(np.std(np.diff(s.values))) if len(s) > 1 else float(np.std(s.values)) |
| lower, upper, ci_detail = empirical_intervals(mean, errors_by_h, level, resid_std) |
|
|
| |
| clipped = [] |
| if use_guardrails: |
| mean, clipped = apply_guardrails(mean, s.values) |
| lower = np.minimum(lower, mean) |
| upper = np.maximum(upper, mean) |
|
|
| labels = next_periods(s.index[-1], periods) |
|
|
| |
| last3 = float(np.mean(s.values[-3:])) |
| fc3 = float(np.mean(mean[:min(3, periods)])) |
| change = (fc3 - last3) / last3 if last3 else 0.0 |
| trend = "up" if change > 0.05 else "down" if change < -0.05 else "stable" |
|
|
| |
| acc = None |
| if bt and used in bt: |
| m = bt[used]["metrics"] |
| acc = { |
| "mape": round(m["mape"], 2), |
| "mdape": round(m["mdape"], 2), |
| "mae": round(m["mae"], 2), |
| "rmse": round(m["rmse"], 2), |
| "method": "rolling-origin backtest", |
| "n_origins": bt_meta.get("n_origins"), |
| "horizon_tested": bt_meta.get("horizon"), |
| "baseline_mape": None if bench_mape is None else round(bench_mape, 2), |
| } |
| acc["reliability"] = reliability_label(m["mape"], used, leaderboard) |
|
|
| explanation = explain_forecast(s, used, mean, labels, leaderboard, acc, |
| ci_detail, quality, level) |
|
|
| |
| confidence = None |
| if acc: |
| confidence = round(max(0.0, 1 - min(acc["mape"], 100) / 100), 3) |
|
|
| |
| agg_mape = None |
| if bt and used in bt: |
| agg_mape = round(bt[used]["metrics_aggregate"]["mape"], 2) |
| total = { |
| "periods": f"{labels[0]} → {labels[-1]}", |
| "predicted": round(float(np.sum(mean)), 2), |
| "lower": round(float(max(0.0, np.sum(mean) - Z[level] * ci_detail["sigma_by_horizon"][-1] * np.sqrt(periods))), 2), |
| "upper": round(float(np.sum(mean) + Z[level] * ci_detail["sigma_by_horizon"][-1] * np.sqrt(periods)), 2), |
| "mape_backtested": agg_mape, |
| "note": ( |
| "ยอดรวมทั้งช่วงมักแม่นกว่ารายเดือน เพราะเดือนที่ทายสูงเกินกับต่ำเกินหักล้างกัน " |
| "— ถ้าจะเอาไปตั้งเป้า ให้ตั้งเป็นยอดรวมไตรมาส/ครึ่งปี ไม่ใช่รายเดือน" |
| ), |
| } |
|
|
| return jsonify({ |
| "forecast": [ |
| {"period": p, "predicted": round(float(mv), 2), |
| "lower": round(float(lo), 2), "upper": round(float(up), 2)} |
| for p, mv, lo, up in zip(labels, mean, lower, upper) |
| ], |
| "total": total, |
| "trend": trend, |
| "trend_change_pct": round(change * 100, 1), |
| "confidence": confidence, |
| "confidence_note": ( |
| "ค่านี้คำนวณจาก MAPE ที่วัดได้จริง (1 − MAPE) ไม่ใช่ 'ความมั่นใจ' ทางสถิติ " |
| "— ให้ดู accuracy.mape เป็นหลัก" |
| ), |
| "accuracy": acc, |
| "explanation": explanation, |
| "leaderboard": leaderboard, |
| "data_quality": quality, |
| "warnings": warns + quality["notes"], |
| "meta": { |
| "used_model": used, |
| "used_model_th": MODEL_INFO.get(used, {}).get("th", used), |
| "selection_reason": selection, |
| "fallback": fallback_used, |
| "n_history": len(s), |
| "history_range": f"{quality['start']} → {quality['end']}", |
| "interval_level": level, |
| "interval_method": ci_detail["method"], |
| "interval_detail": ci_detail, |
| "guardrails": use_guardrails, |
| "guardrails_clipped_periods": [labels[i] for i in clipped], |
| "backtest": bt_meta if isinstance(bt_meta, dict) else {"note": bt_meta}, |
| "version": "2.2-teaching", |
| }, |
| }) |
| except Exception as e: |
| return jsonify({"error": f"{type(e).__name__}: {e}"}), 500 |
|
|
|
|
| @app.route("/api/backtest", methods=["POST", "OPTIONS"]) |
| def backtest_endpoint(): |
| """ทดสอบย้อนหลังอย่างเดียว — สำหรับหน้าจอ 'เทียบโมเดล' ในห้องเรียน""" |
| if request.method == "OPTIONS": |
| return ("", 204) |
| try: |
| body = request.get_json(force=True) or {} |
| rows = body.get("data", []) |
| horizon = max(1, min(12, int(body.get("horizon", BACKTEST_HORIZON)))) |
| |
| tw = body.get("train_window") |
| tw = int(tw) if tw not in (None, "", "expanding") else None |
| if tw is not None: |
| tw = max(SEASONAL_PERIOD + 3, min(240, tw)) |
| logt = bool(body.get("log_transform", False)) |
|
|
| if len(rows) < MIN_HISTORY: |
| return jsonify({"error": f"ต้องการข้อมูลอย่างน้อย {MIN_HISTORY} เดือน"}), 400 |
|
|
| s, warns = to_series(rows) |
| if tw is None and not logt: |
| bt, meta = _cached_backtest(rows, s, horizon) |
| else: |
| key = _hash(rows, f"|bt{horizon}|w{tw}|l{int(logt)}") |
| if key in _CACHE: |
| bt, meta = _CACHE[key] |
| else: |
| if logt and float(np.min(s.values)) <= 0: |
| return jsonify({"error": "log_transform ใช้ไม่ได้เมื่อมีเดือนที่ยอด ≤ 0"}), 400 |
| bt, meta = rolling_origin_backtest(s, horizon=horizon, train_window=tw, |
| log_transform=logt) |
| if len(_CACHE) >= _CACHE_MAX: |
| _CACHE.pop(next(iter(_CACHE))) |
| _CACHE[key] = (bt, meta) |
| warns.append( |
| "กำลังใช้ค่าตั้งแบบทดลอง (train_window / log_transform) — " |
| "การไล่ลองหลายค่าแล้วเลือกตัวที่ MAPE ต่ำสุดจากชุดทดสอบชุดเดียวกัน " |
| "คือการ overfit ขั้นตอนเลือกค่าตั้ง ตัวเลขที่ได้จะดีเกินจริง" |
| ) |
| if not bt: |
| return jsonify({"error": meta if isinstance(meta, str) else "ทดสอบย้อนหลังไม่ได้"}), 400 |
|
|
| leaderboard, bench = build_leaderboard(bt) |
|
|
| |
| by_h = {} |
| for name, d in bt.items(): |
| by_h[name] = { |
| str(h): round(float(np.mean(np.abs(e))), 2) |
| for h, e in d["errors_by_h"].items() if e |
| } |
|
|
| monthly_vs_total = [ |
| {"model": r["model"], "label": r["label"], |
| "mape_monthly": r["mape"], "mape_total": r["mape_total"], |
| "improvement_pct_points": round(r["mape"] - r["mape_total"], 2)} |
| for r in leaderboard |
| ] |
|
|
| return jsonify({ |
| "leaderboard": leaderboard, |
| "monthly_vs_total": monthly_vs_total, |
| "mae_by_horizon": by_h, |
| "benchmark": {"model": BENCHMARK, "mape": None if bench is None else round(bench, 2)}, |
| "experiment": { |
| "train_window": tw or "expanding", |
| "log_transform": logt, |
| "is_default": tw is None and not logt, |
| "how_to_check_honestly": ( |
| "ถ้าจะเลือกค่าตั้งจริง ต้องแบ่งเวลาออกเป็นสองช่วง: ใช้ช่วงแรกเลือกค่าตั้ง " |
| "แล้วเอาค่าที่เลือกไปวัดกับช่วงหลังที่ยังไม่เคยเห็น " |
| "ถ้าค่าตั้งที่ชนะในช่วงแรกไม่ชนะในช่วงหลัง แปลว่าที่ชนะคือความบังเอิญ" |
| ), |
| }, |
| "interpretation": reliability_label(leaderboard[0]["mape"], leaderboard[0]["model"], leaderboard), |
| "warnings": warns, |
| "meta": meta, |
| }) |
| except Exception as e: |
| return jsonify({"error": f"{type(e).__name__}: {e}"}), 500 |
|
|
|
|
| @app.route("/api/decompose", methods=["POST", "OPTIONS"]) |
| def decompose(): |
| """แยกอนุกรมเวลาเป็น แนวโน้ม + ฤดูกาล + ส่วนที่เหลือ (STL) |
| |
| ค่าความแรง (strength) ตามนิยามของ Hyndman: |
| strength = max(0, 1 - Var(residual) / Var(residual + component)) |
| > 0.6 ถือว่าองค์ประกอบนั้นเด่นชัด |
| """ |
| if request.method == "OPTIONS": |
| return ("", 204) |
| try: |
| from statsmodels.tsa.seasonal import STL |
|
|
| body = request.get_json(force=True) or {} |
| rows = body.get("data", []) |
| if len(rows) < 2 * SEASONAL_PERIOD: |
| return jsonify({"error": f"ต้องการอย่างน้อย {2 * SEASONAL_PERIOD} เดือนจึงจะแยกฤดูกาลได้"}), 400 |
|
|
| s, warns = to_series(rows) |
| res = STL(s, period=SEASONAL_PERIOD, robust=True).fit() |
| tr, se, rs = res.trend, res.seasonal, res.resid |
|
|
| def strength(comp): |
| v_r, v_rc = np.var(rs), np.var(rs + comp) |
| return round(float(max(0.0, 1 - v_r / v_rc)) if v_rc > 0 else 0.0, 3) |
|
|
| st_trend, st_seas = strength(tr), strength(se) |
|
|
| |
| seas_idx = pd.Series(se.values, index=s.index.month).groupby(level=0).mean() |
| month_th = ["ม.ค.", "ก.พ.", "มี.ค.", "เม.ย.", "พ.ค.", "มิ.ย.", |
| "ก.ค.", "ส.ค.", "ก.ย.", "ต.ค.", "พ.ย.", "ธ.ค."] |
|
|
| reading = [] |
| reading.append( |
| f"ความแรงของแนวโน้ม {st_trend} — " + |
| ("แนวโน้มชัดเจน" if st_trend > 0.6 else "แนวโน้มอ่อน ข้อมูลแกว่งรอบค่าเฉลี่ยเป็นหลัก") |
| ) |
| reading.append( |
| f"ความแรงของฤดูกาล {st_seas} — " + |
| ("รูปแบบรายเดือนซ้ำได้ชัด ใช้โมเดลฤดูกาลคุ้ม" |
| if st_seas > 0.6 else |
| "รูปแบบรายเดือนไม่ซ้ำพอ — โมเดลฤดูกาลอาจไม่ช่วย และอาจ overfit") |
| ) |
| var_share = float(np.var(rs) / np.var(s.values)) if np.var(s.values) > 0 else 0 |
| reading.append( |
| f"ส่วนที่อธิบายไม่ได้ (residual) กินความผันผวนไป {var_share*100:.0f}% " |
| + ("— นี่คือเพดานของการพยากรณ์ด้วยข้อมูลชุดนี้" if var_share > 0.5 else "") |
| ) |
|
|
| return jsonify({ |
| "periods": [d.strftime("%Y-%m") for d in s.index], |
| "observed": [round(float(x), 2) for x in s.values], |
| "trend": [round(float(x), 2) for x in tr.values], |
| "seasonal": [round(float(x), 2) for x in se.values], |
| "residual": [round(float(x), 2) for x in rs.values], |
| "strength": {"trend": st_trend, "seasonal": st_seas}, |
| "seasonal_index": [ |
| {"month": int(m), "month_th": month_th[int(m) - 1], "effect": round(float(v), 2)} |
| for m, v in seas_idx.items() |
| ], |
| "residual_variance_share": round(var_share, 3), |
| "reading": reading, |
| "warnings": warns, |
| }) |
| except Exception as e: |
| return jsonify({"error": f"{type(e).__name__}: {e}"}), 500 |
|
|
|
|
| @app.route("/api/diagnostics", methods=["POST", "OPTIONS"]) |
| def diagnostics(): |
| """ADF stationarity + ACF/PACF + คุณภาพข้อมูล — ใช้ก่อนเลือกพารามิเตอร์ SARIMA""" |
| if request.method == "OPTIONS": |
| return ("", 204) |
| try: |
| from statsmodels.tsa.stattools import acf, adfuller, pacf |
|
|
| body = request.get_json(force=True) or {} |
| rows = body.get("data", []) |
| nlags = max(6, min(24, int(body.get("nlags", 24)))) |
| if len(rows) < MIN_HISTORY: |
| return jsonify({"error": f"ต้องการข้อมูลอย่างน้อย {MIN_HISTORY} เดือน"}), 400 |
|
|
| s, warns = to_series(rows) |
| v = s.values.astype(float) |
| nlags = min(nlags, len(v) // 2 - 1) |
|
|
| def adf(x, label): |
| stat, p, lags, nobs, crit, _ = adfuller(x, autolag="AIC") |
| return { |
| "series": label, |
| "statistic": round(float(stat), 4), |
| "p_value": round(float(p), 4), |
| "lags_used": int(lags), |
| "critical_5pct": round(float(crit["5%"]), 4), |
| "is_stationary": bool(p < 0.05), |
| "reading": ("นิ่งแล้ว (p < 0.05) — ไม่ต้อง differencing เพิ่ม" |
| if p < 0.05 else |
| "ยังไม่นิ่ง (p ≥ 0.05) — ควร differencing"), |
| } |
|
|
| adf_raw = adf(v, "ข้อมูลดิบ") |
| adf_d1 = adf(np.diff(v), "หลัง differencing 1 ครั้ง") if len(v) > 3 else None |
|
|
| a = acf(v, nlags=nlags, fft=False) |
| p_ = pacf(v, nlags=min(nlags, len(v) // 2 - 1)) |
| bound = 1.96 / np.sqrt(len(v)) |
|
|
| sig_seasonal = [ |
| int(k) for k in [12, 24] if k <= nlags and abs(float(a[k])) > bound |
| ] |
|
|
| hints = [] |
| d_sug = 0 if adf_raw["is_stationary"] else 1 |
| hints.append(f"แนะนำ d = {d_sug} (จากผล ADF)") |
| if sig_seasonal: |
| hints.append(f"ACF ที่ lag {sig_seasonal} เกินเส้นนัยสำคัญ — มีสัญญาณฤดูกาล 12 เดือน ควรตั้ง D = 1") |
| else: |
| hints.append("ACF ที่ lag 12/24 ไม่เกินเส้นนัยสำคัญ — สัญญาณฤดูกาลอ่อน อาจไม่ต้องใส่ส่วนฤดูกาล") |
| sig_early = [int(k) for k in range(1, min(4, nlags + 1)) if abs(float(p_[k])) > bound] |
| hints.append( |
| f"PACF ตัดหลัง lag {max(sig_early)} — ลอง p = {max(sig_early)}" |
| if sig_early else "PACF ไม่มี lag ต้น ๆ ที่เด่น — ลอง p = 0 หรือ 1" |
| ) |
|
|
| return jsonify({ |
| "adf": [x for x in [adf_raw, adf_d1] if x], |
| "acf": [{"lag": i, "value": round(float(a[i]), 4), "significant": bool(abs(float(a[i])) > bound)} |
| for i in range(len(a))], |
| "pacf": [{"lag": i, "value": round(float(p_[i]), 4), "significant": bool(abs(float(p_[i])) > bound)} |
| for i in range(len(p_))], |
| "significance_bound": round(float(bound), 4), |
| "sarima_hints": hints, |
| "data_quality": data_quality(s), |
| "warnings": warns, |
| }) |
| except Exception as e: |
| return jsonify({"error": f"{type(e).__name__}: {e}"}), 500 |
|
|
|
|
| |
| |
| |
|
|
| def _year_table(s): |
| """ตารางปี × เดือน พร้อมธงบอกว่าปีไหน/เดือนไหนข้อมูลยังไม่ครบ""" |
| df = pd.DataFrame({"v": s.values, "y": s.index.year, "m": s.index.month}) |
| piv = df.pivot_table(index="y", columns="m", values="v", aggfunc="sum") |
| piv = piv.reindex(columns=range(1, 13)) |
| return piv |
|
|
|
|
| def _nn(x): |
| """แปลง NaN เป็น None เพื่อให้ JSON อ่านได้""" |
| return None if x is None or (isinstance(x, float) and not np.isfinite(x)) else round(float(x), 2) |
|
|
|
|
| @app.route("/api/charts", methods=["POST", "OPTIONS"]) |
| def charts(): |
| """ชุดข้อมูลพร้อมวาดกราฟหลายแบบจากอนุกรมเดียวกัน |
| |
| คืนเป็น "ข้อมูลที่พร้อมวาด" ทุกบล็อก — หน้าเว็บเลือกหยิบไปวาดได้เลย |
| ทุกบล็อกที่อิงรายปีจะมีธง complete บอกว่าปีนั้นข้อมูลครบ 12 เดือนหรือยัง |
| เพื่อไม่ให้กราฟหลอกตาว่า "ปีนี้ยอดตก" ทั้งที่ยังไม่จบปี |
| """ |
| if request.method == "OPTIONS": |
| return ("", 204) |
| try: |
| body = request.get_json(force=True) or {} |
| rows = body.get("data", []) |
| drop_last = bool(body.get("drop_last_incomplete", False)) |
| if len(rows) < 3: |
| return jsonify({"error": "ต้องการข้อมูลอย่างน้อย 3 เดือน"}), 400 |
|
|
| s, warns = to_series(rows) |
| if drop_last and len(s) > 3: |
| warns.append(f"ตัดเดือนสุดท้าย ({s.index[-1].strftime('%Y-%m')}) ออกตามที่ร้องขอ") |
| s = s.iloc[:-1] |
|
|
| piv = _year_table(s) |
| years = [int(y) for y in piv.index] |
| month_th = ["ม.ค.", "ก.พ.", "มี.ค.", "เม.ย.", "พ.ค.", "มิ.ย.", |
| "ก.ค.", "ส.ค.", "ก.ย.", "ต.ค.", "พ.ย.", "ธ.ค."] |
|
|
| |
| totals, counts = piv.sum(axis=1), piv.notna().sum(axis=1) |
| yoy_annual = [] |
| prev = None |
| for i, y in enumerate(years): |
| tot, cnt = float(totals.iloc[i]), int(counts.iloc[i]) |
| growth = None if prev in (None, 0) else round((tot - prev) / prev * 100, 2) |
| yoy_annual.append({ |
| "year": y, "total": round(tot, 2), "months_with_data": cnt, |
| "complete": cnt == 12, "growth_pct": growth, |
| "avg_per_month": round(tot / cnt, 2) if cnt else None, |
| }) |
| prev = tot |
|
|
| |
| yoy_monthly = { |
| "months": month_th, |
| "series": [{"year": int(y), |
| "values": [_nn(piv.loc[y, m]) for m in range(1, 13)]} |
| for y in years], |
| } |
|
|
| |
| ytd = {"months": month_th, "series": []} |
| for y in years: |
| run, acc = [], 0.0 |
| for m in range(1, 13): |
| v = piv.loc[y, m] |
| if pd.isna(v): |
| run.append(None) |
| else: |
| acc += float(v) |
| run.append(round(acc, 2)) |
| ytd["series"].append({"year": int(y), "values": run}) |
|
|
| |
| heat = [] |
| for y in years: |
| row = piv.loc[y] |
| mean = row.mean() |
| heat.append({"year": int(y), |
| "values": [None if pd.isna(row[m]) or not mean else round(float(row[m]) / mean, 3) |
| for m in range(1, 13)]}) |
|
|
| |
| ma = { |
| "periods": [d.strftime("%Y-%m") for d in s.index], |
| "actual": [round(float(x), 2) for x in s.values], |
| "ma3": [_nn(x) for x in s.rolling(3).mean().values], |
| "ma12": [_nn(x) for x in s.rolling(12).mean().values], |
| } |
|
|
| |
| mom = s.pct_change() * 100 |
| yoy_m = s.pct_change(12) * 100 |
| growth = { |
| "periods": ma["periods"], |
| "mom_pct": [_nn(x) for x in mom.values], |
| "yoy_pct": [_nn(x) for x in yoy_m.values], |
| } |
|
|
| |
| cnts, edges = np.histogram(s.values, bins=min(12, max(5, len(s) // 7))) |
| dist = { |
| "bins": [{"from": round(float(edges[i]), 2), "to": round(float(edges[i + 1]), 2), |
| "count": int(cnts[i])} for i in range(len(cnts))], |
| "mean": round(float(s.mean()), 2), "median": round(float(s.median()), 2), |
| } |
|
|
| |
| prof = [] |
| norm = pd.DataFrame({"v": s.values, "y": s.index.year, "m": s.index.month}) |
| norm["idx"] = norm.groupby("y")["v"].transform(lambda x: x / x.mean()) |
| for m in range(1, 13): |
| g = norm.loc[norm.m == m, "idx"].dropna() |
| if len(g) == 0: |
| continue |
| q1, q3 = float(g.quantile(.25)), float(g.quantile(.75)) |
| prof.append({"month": m, "month_th": month_th[m - 1], "n": int(len(g)), |
| "min": round(float(g.min()), 3), "q1": round(q1, 3), |
| "median": round(float(g.median()), 3), "q3": round(q3, 3), |
| "max": round(float(g.max()), 3), "iqr": round(q3 - q1, 3), |
| "repeatable": bool(q3 - q1 < 0.35)}) |
|
|
| |
| by_category = None |
| cats = [r for r in rows if isinstance(r, dict) and r.get("category")] |
| if cats: |
| cdf = pd.DataFrame(cats) |
| cdf["sales"] = pd.to_numeric(cdf["sales"], errors="coerce").fillna(0) |
| cdf["ym"] = cdf["date"].astype(str).str.strip() |
| tot = cdf.groupby("category")["sales"].sum().sort_values(ascending=False) |
| share = (tot / tot.sum() * 100).round(2) |
| by_category = { |
| "totals": [{"category": str(k), "total": round(float(v), 2), |
| "share_pct": float(share[k])} for k, v in tot.items()], |
| "monthly": [{"category": str(c), |
| "periods": sorted(g["ym"].unique().tolist()), |
| "values": [round(float(g.loc[g.ym == p, "sales"].sum()), 2) |
| for p in sorted(g["ym"].unique().tolist())]} |
| for c, g in cdf.groupby("category")], |
| "note": ("เกิน 3 หมวดให้ใช้ bar เรียงลำดับ ไม่ใช่ pie " |
| "— ตาคนเทียบความยาวแท่งได้แม่นกว่ามุมของชิ้นพาย"), |
| } |
|
|
| partial = [r["year"] for r in yoy_annual if not r["complete"]] |
| if partial: |
| warns.append( |
| f"ปี {partial} ยังมีข้อมูลไม่ครบ 12 เดือน — ห้ามเทียบยอดรวมรายปีตรง ๆ " |
| "ให้ใช้ค่าเฉลี่ยต่อเดือนหรือ YTD แทน" |
| ) |
|
|
| return jsonify({ |
| "yoy_annual": yoy_annual, |
| "yoy_monthly": yoy_monthly, |
| "ytd": ytd, |
| "heatmap": heat, |
| "moving_average": ma, |
| "growth": growth, |
| "distribution": dist, |
| "month_profile": prof, |
| "by_category": by_category, |
| "data_quality": data_quality(s), |
| "warnings": warns + data_quality(s)["notes"], |
| "chart_guide": { |
| "yoy_annual": "bar — เทียบยอดรวมรายปี (ระวังปีที่ยังไม่ครบ)", |
| "yoy_monthly": "line หลายเส้น หรือ grouped bar — เทียบเดือนเดียวกันข้ามปี", |
| "ytd": "line — เทียบ 'จังหวะ' การสะสมยอด ปีนี้เร็วกว่าปีที่แล้วไหม", |
| "heatmap": "heatmap ปี × เดือน — มองหาคอลัมน์ที่เข้ม/อ่อนทั้งแนวตั้ง", |
| "moving_average": "line — เส้นจริง + เส้นเรียบ 3 และ 12 เดือน", |
| "growth": "bar มีค่าบวกลบ — MoM และ YoY รายเดือน", |
| "distribution": "histogram — ยอดรายเดือนกระจายตัวอย่างไร", |
| "month_profile": "box plot — ฤดูกาลซ้ำจริงหรือแค่บังเอิญปีเดียว", |
| }, |
| }) |
| except Exception as e: |
| return jsonify({"error": f"{type(e).__name__}: {e}"}), 500 |
|
|
|
|
| @app.route("/api/hindcast", methods=["POST", "OPTIONS"]) |
| def hindcast(): |
| """ทำนายย้อนกลับ — ให้โมเดลทายอดีตทีละเดือน โดยเห็นเฉพาะข้อมูลก่อนหน้า |
| |
| ต่างจาก /api/backtest ตรงที่คืน "เส้นกราฟ" ให้วาดทับกับของจริงได้ |
| ทุกค่าที่คืนเป็น out-of-sample ล้วน ไม่มีการมองอนาคตแม้แต่จุดเดียว |
| |
| และตอบคำถามสำคัญ: อัตราเติบโต YoY ที่โมเดลทำนาย |
| สอดคล้องกับ YoY ที่เกิดขึ้นจริงหรือไม่ |
| """ |
| if request.method == "OPTIONS": |
| return ("", 204) |
| try: |
| body = request.get_json(force=True) or {} |
| rows = body.get("data", []) |
| horizon = max(1, min(6, int(body.get("horizon", 1)))) |
| drop_last = bool(body.get("drop_last_incomplete", False)) |
| want = body.get("models") or ["mean3", "seasonal_naive", "linear", "holtwinters", "ml_rf"] |
| want = [ALIASES.get(str(m).lower(), str(m).lower()) for m in want] |
| want = [m for m in want if m in MODELS] |
| if not want: |
| return jsonify({"error": "ไม่มีชื่อโมเดลที่ใช้ได้"}), 400 |
|
|
| s, warns = to_series(rows) |
| if drop_last and len(s) > MIN_HISTORY: |
| warns.append(f"ตัดเดือนสุดท้าย ({s.index[-1].strftime('%Y-%m')}) ออกตามที่ร้องขอ") |
| s = s.iloc[:-1] |
|
|
| n = len(s) |
| min_train = max(MIN_TRAIN, SEASONAL_PERIOD + 3) |
| if n < min_train + horizon + 2: |
| min_train = max(SEASONAL_PERIOD + 3, n - horizon - 6) |
| if n < min_train + horizon: |
| return jsonify({"error": f"ต้องการข้อมูลอย่างน้อย {min_train + horizon} เดือน"}), 400 |
|
|
| periods = [d.strftime("%Y-%m") for d in s.index] |
|
|
| ck = _hash(rows, f"|hc{horizon}|{drop_last}|{','.join(sorted(want))}") |
| if ck in _CACHE: |
| return jsonify(_CACHE[ck]) |
|
|
| preds = {m: [None] * n for m in want} |
|
|
| t0 = time.time() |
| for o in range(min_train, n - horizon + 1): |
| train = s.iloc[:o] |
| for m in want: |
| try: |
| fc = MODELS[m](train, horizon) |
| if np.all(np.isfinite(fc)): |
| preds[m][o + horizon - 1] = round(float(fc[horizon - 1]), 2) |
| except Exception: |
| pass |
| elapsed = round(time.time() - t0, 2) |
|
|
| actual = [round(float(x), 2) for x in s.values] |
|
|
| |
| summary = {} |
| for m in want: |
| pairs = [(a, p) for a, p in zip(actual, preds[m]) if p is not None] |
| if not pairs: |
| continue |
| a = np.array([x[0] for x in pairs], dtype=float) |
| p = np.array([x[1] for x in pairs], dtype=float) |
| sc = score(a, p) |
| summary[m] = { |
| "label": MODEL_INFO.get(m, {}).get("th", m), |
| "kind": MODEL_INFO.get(m, {}).get("kind", "model"), |
| "n_points": len(pairs), |
| "mape": round(sc["mape"], 2), "mae": round(sc["mae"], 2), |
| "rmse": round(sc["rmse"], 2), |
| "bias": round(float(np.mean(p - a)), 2), |
| } |
|
|
| |
| ydf = pd.DataFrame({"period": periods, "actual": actual}) |
| ydf["year"] = [int(str(p)[:4]) for p in periods] |
| for m in want: |
| ydf[m] = preds[m] |
|
|
| yoy_check = [] |
| yrs = sorted(ydf["year"].unique()) |
| for i in range(1, len(yrs)): |
| cur, prv = yrs[i], yrs[i - 1] |
| rows_cur = ydf[ydf.year == cur] |
| rows_prv = ydf[ydf.year == prv] |
| |
| entry = {"year": int(cur), "vs_year": int(prv), "models": {}} |
| for m in want: |
| cm = rows_cur.dropna(subset=[m]) |
| if len(cm) < 3: |
| continue |
| months = set(cm["period"].str[5:]) |
| pm = rows_prv[rows_prv["period"].str[5:].isin(months)] |
| if len(pm) < 3: |
| continue |
| a_now, a_prev = float(cm["actual"].sum()), float(pm["actual"].sum()) |
| p_now = float(cm[m].sum()) |
| if a_prev == 0: |
| continue |
| g_act = (a_now - a_prev) / a_prev * 100 |
| g_pred = (p_now - a_prev) / a_prev * 100 |
| entry["actual_growth_pct"] = round(float(g_act), 2) |
| entry["months_compared"] = int(len(cm)) |
| entry["models"][m] = { |
| "predicted_growth_pct": round(float(g_pred), 2), |
| "gap_pct_points": round(float(g_pred - g_act), 2), |
| "same_direction": bool((g_act >= 0) == (g_pred >= 0)), |
| } |
| if entry["models"]: |
| yoy_check.append(entry) |
|
|
| |
| reading = [] |
| if summary: |
| best = min(summary, key=lambda m: summary[m]["mape"]) |
| reading.append( |
| f"โมเดลที่ทายย้อนหลังได้ใกล้ความจริงที่สุดคือ {summary[best]['label']} " |
| f"(MAPE {summary[best]['mape']}%)" |
| ) |
| biased = [m for m, v in summary.items() if abs(v["bias"]) > 0.1 * float(np.mean(actual))] |
| if biased: |
| reading.append( |
| "โมเดลที่ทายเบี่ยงไปทางเดียวอย่างเป็นระบบ (bias เกิน 10% ของค่าเฉลี่ย): " |
| + ", ".join(summary[m]["label"] for m in biased) |
| + " — bias แก้ได้ด้วยการปรับค่าคงที่ ต่างจากความผิดพลาดแบบสุ่มที่แก้ไม่ได้" |
| ) |
| if yoy_check: |
| ok = sum(1 for e in yoy_check for v in e["models"].values() if v["same_direction"]) |
| tot = sum(len(e["models"]) for e in yoy_check) |
| reading.append( |
| f"ทิศทางการเติบโต YoY: โมเดลทายถูกทิศ {ok} จาก {tot} ครั้ง " |
| + ("— ใช้บอกทิศทางได้ แม้ตัวเลขจะไม่ตรง" |
| if tot and ok / tot >= 0.6 else |
| "— ยังบอกแม้แต่ทิศทางไม่ได้อย่างน่าเชื่อถือ") |
| ) |
| reading.append( |
| "ทุกค่าในกราฟนี้เป็น out-of-sample ล้วน — ณ แต่ละจุด โมเดลเห็นเฉพาะข้อมูลก่อนหน้านั้น" |
| ) |
|
|
| payload = { |
| "periods": periods, |
| "actual": actual, |
| "predicted": preds, |
| "summary": summary, |
| "yoy_check": yoy_check, |
| "reading": reading, |
| "warnings": warns, |
| "meta": { |
| "horizon": horizon, |
| "min_train_months": min_train, |
| "first_predicted_period": periods[min_train + horizon - 1], |
| "models": want, |
| "elapsed_sec": elapsed, |
| "note": ("จุดก่อนหน้า first_predicted_period เป็น null เพราะยังไม่มีข้อมูลพอเทรน " |
| "— ไม่ใช่ข้อผิดพลาด"), |
| }, |
| } |
| if len(_CACHE) >= _CACHE_MAX: |
| _CACHE.pop(next(iter(_CACHE))) |
| _CACHE[ck] = payload |
| return jsonify(payload) |
| except Exception as e: |
| return jsonify({"error": f"{type(e).__name__}: {e}"}), 500 |
|
|
|
|
|
|
| @app.route("/api/describe", methods=["POST", "OPTIONS"]) |
| def describe(): |
| """อธิบายข้อมูลชุดนี้ให้คนที่ไม่เคยวิเคราะห์ข้อมูลมาก่อนเข้าใจ |
| |
| ตอบ 4 คำถามที่ผู้เริ่มต้นถามเสมอ: |
| 1. ข้อมูลนี้คืออะไร จากไหน ถึงไหน |
| 2. แต่ละคอลัมน์หมายถึงอะไร |
| 3. ข้อมูลพร้อมใช้หรือยัง ติดอะไรบ้าง |
| 4. แล้วต้องทำอะไรต่อ ตามลำดับไหน |
| """ |
| if request.method == "OPTIONS": |
| return ("", 204) |
| try: |
| body = request.get_json(force=True) or {} |
| rows = body.get("data", []) |
| unit = str(body.get("unit", "บาท")) |
| subject = str(body.get("subject", "ยอดขาย")) |
| if len(rows) < 2: |
| return jsonify({"error": "ต้องการข้อมูลอย่างน้อย 2 เดือน"}), 400 |
|
|
| s, warns = to_series(rows) |
| q = data_quality(s) |
| v = np.asarray(s.values, dtype=float) |
| years = sorted({d.year for d in s.index}) |
|
|
| |
| overview = [ |
| f"ข้อมูลชุดนี้คือ{subject} สรุปเป็น รายเดือน เดือนละ 1 ค่า", |
| f"ครอบคลุมตั้งแต่ {q['start']} ถึง {q['end']} รวม {q['n_months']} เดือน " |
| f"({len(years)} ปี: {years[0]}–{years[-1]})", |
| f"ค่าเฉลี่ยเดือนละ {q['mean']:,.0f} {unit} · ต่ำสุด {q['min']:,.0f} · สูงสุด {q['max']:,.0f}", |
| f"หนึ่งแถว = หนึ่งเดือน ไม่ใช่หนึ่งรายการขาย — รายการขายทั้งหมดในเดือนนั้นถูกบวกรวมกันแล้ว", |
| ] |
|
|
| |
| dictionary = [ |
| {"column": "date", "th": "เดือน", "type": "ข้อความ รูปแบบ YYYY-MM", |
| "example": q["start"], "meaning": "เดือนที่ข้อมูลนั้นเกิดขึ้น เช่น 2025-03 คือ มีนาคม 2025"}, |
| {"column": "sales", "th": f"{subject} ({unit})", "type": "ตัวเลข", |
| "example": f"{v[0]:,.0f}", "meaning": f"ยอดรวมทั้งเดือน หน่วยเป็น{unit}"}, |
| ] |
| if any(isinstance(r, dict) and r.get("category") for r in rows): |
| dictionary.append({"column": "category", "th": "หมวดสินค้า", "type": "ข้อความ", |
| "example": "Computer/PC", |
| "meaning": "กลุ่มสินค้า ใช้แยกดูว่าหมวดไหนโตหรือหด (ไม่บังคับ)"}) |
|
|
| |
| def peek(idx): |
| return [{"date": s.index[i].strftime("%Y-%m"), "sales": round(float(v[i]), 2)} for i in idx] |
| preview = { |
| "first": peek(range(min(5, len(v)))), |
| "last": peek(range(max(0, len(v) - 5), len(v))), |
| "note": "แสดง 5 แถวแรกและ 5 แถวสุดท้าย จากทั้งหมด %d แถว" % len(v), |
| } |
|
|
| |
| def item(label, ok, detail): |
| return {"check": label, "status": "ok" if ok else "warn", "detail": detail} |
| full = pd.period_range(s.index.min().to_period("M"), s.index.max().to_period("M"), freq="M") |
| checklist = [ |
| item("เดือนต่อเนื่องไม่ขาด", len(full) == q["n_months"], |
| "ครบทุกเดือน" if len(full) == q["n_months"] |
| else f"ควรมี {len(full)} เดือน แต่มีจริง {q['n_months']} เดือน"), |
| item("ไม่มีเดือนที่ยอดเป็นศูนย์", q["zero_months"] == 0, |
| "ไม่มี" if q["zero_months"] == 0 else f"พบ {q['zero_months']} เดือน — ตรวจว่าไม่ได้ขายจริง หรือลืมบันทึก"), |
| item("ยาวพอฝึกโมเดลฤดูกาล", q["n_months"] >= MIN_TRAIN, |
| f"{q['n_months']} เดือน (ต้องการ ≥ {MIN_TRAIN})"), |
| item("ความผันผวนอยู่ในระดับที่พยากรณ์ได้ดี", q["cv"] <= 0.35, |
| f"CV = {q['cv']}" + ("" if q["cv"] <= 0.35 else " — สูง คาดหวังความผิดพลาดหลายสิบเปอร์เซ็นต์")), |
| item("ไม่มีเดือนค่าสุดโต่ง", not q["outliers"], |
| "ไม่มี" if not q["outliers"] |
| else "พบ " + ", ".join(o["period"] for o in q["outliers"]) + " — ตรวจว่าเป็นดีลจริงหรือข้อมูลผิด"), |
| item("เดือนสุดท้ายเก็บข้อมูลครบ", |
| not any("เดือนล่าสุด" in n for n in q["notes"]), |
| "ดูปกติ" if not any("เดือนล่าสุด" in n for n in q["notes"]) |
| else f"{q['end']} ต่ำผิดปกติ — น่าจะยังเก็บไม่ครบเดือน ควรตัดออกก่อนฝึกโมเดล"), |
| ] |
|
|
| |
| plan = [ |
| {"step": 1, "title": "ทำความรู้จักข้อมูล", |
| "question": "ข้อมูลนี้คืออะไร ครบไหม เชื่อได้ไหม", |
| "endpoint": "POST /api/describe", |
| "look_at": "เช็คลิสต์ด้านบน — ทุกข้อที่ขึ้นเตือน ต้องตัดสินใจก่อนไปต่อ", |
| "why": "ข้อมูลเสียทำให้ทุกอย่างหลังจากนี้ผิดหมด และผิดแบบดูไม่ออก"}, |
| {"step": 2, "title": "ดูภาพรวมด้วยกราฟ", |
| "question": "ยอดขึ้นลงอย่างไร ปีไหนดี เดือนไหนแรง", |
| "endpoint": "POST /api/charts", |
| "look_at": "เริ่มที่ yoy_annual (ยอดรายปี) → yoy_monthly (เทียบเดือนข้ามปี) → heatmap", |
| "why": "ตาเห็นรูปแบบก่อน แล้วค่อยเอาตัวเลขมายืนยัน ไม่ใช่กลับกัน"}, |
| {"step": 3, "title": "ตรวจว่าพยากรณ์ได้แค่ไหน", |
| "question": "ข้อมูลนี้มีแนวโน้มและฤดูกาลจริงหรือเปล่า", |
| "endpoint": "POST /api/decompose และ POST /api/diagnostics", |
| "look_at": "ความแรงฤดูกาล (>0.6 = ชัด) และสัดส่วน residual (ยิ่งสูง ยิ่งพยากรณ์ยาก)", |
| "why": "รู้เพดานก่อนลงมือ ประหยัดเวลาลองผิดลองถูกได้ทั้งวัน"}, |
| {"step": 4, "title": "เทียบโมเดลกับวิธีเดาแบบง่าย", |
| "question": "โมเดลที่จะใช้ ดีกว่าการเดาง่าย ๆ จริงไหม", |
| "endpoint": "POST /api/backtest", |
| "look_at": "คอลัมน์ skill — ต้องเป็นบวก ถ้าติดลบแปลว่าโมเดลไม่มีประโยชน์", |
| "why": "MAPE 27% จะดีหรือแย่ ตอบไม่ได้จนกว่าจะรู้ว่าวิธีเดาง่าย ๆ ได้เท่าไหร่"}, |
| {"step": 5, "title": "ทำนายย้อนกลับดูของจริง", |
| "question": "ถ้าย้อนเวลาไปใช้โมเดลนี้จริง มันจะทายถูกแค่ไหน", |
| "endpoint": "POST /api/hindcast", |
| "look_at": "เส้นทำนายทับเส้นจริง และตาราง yoy_check ว่าทายถูกทิศหรือไม่", |
| "why": "เห็นกับตาว่าโมเดลพลาดตรงไหน น่าเชื่อถือกว่าตัวเลข MAPE ตัวเดียว"}, |
| {"step": 6, "title": "พยากรณ์อนาคต และอ่านผลให้เป็น", |
| "question": "เดือนหน้าเป็นอย่างไร และเชื่อได้แค่ไหน", |
| "endpoint": "POST /api/forecast", |
| "look_at": "explanation (ทำไมได้ผลแบบนี้) · ช่วงความเชื่อมั่น · total (ยอดรวมทั้งช่วง)", |
| "why": "ตัวเลขเดียวไม่ช่วยตัดสินใจ ช่วง + เหตุผล + ข้อจำกัด คือสิ่งที่ช่วย"}, |
| ] |
|
|
| return jsonify({ |
| "overview": overview, |
| "dictionary": dictionary, |
| "preview": preview, |
| "checklist": checklist, |
| "analysis_plan": plan, |
| "summary_table": { |
| "จำนวนเดือน": q["n_months"], |
| "ช่วงเวลา": f"{q['start']} → {q['end']}", |
| "จำนวนปี": len(years), |
| f"ค่าเฉลี่ยต่อเดือน ({unit})": q["mean"], |
| f"ค่ากลาง ({unit})": q["median"], |
| f"ต่ำสุด ({unit})": q["min"], |
| f"สูงสุด ({unit})": q["max"], |
| "ความผันผวน (CV)": q["cv"], |
| "เดือนที่ขาดหาย": len(full) - q["n_months"], |
| "เดือนค่าสุดโต่ง": len(q["outliers"]), |
| }, |
| "data_quality": q, |
| "warnings": warns + q["notes"], |
| }) |
| except Exception as e: |
| return jsonify({"error": f"{type(e).__name__}: {e}"}), 500 |
|
|
|
|
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
|
|
| ROLES = { |
| "strategic": { |
| "th": "เชิงกลยุทธ์", "en": "Strategic", |
| "user": "ผู้บริหารระดับสูง (CEO / CFO / กรรมการ)", |
| "sees": "ภาพรวมทั้งองค์กร ทิศทางระยะยาว", |
| "cadence": "ทบทวนรอบเดือน – ไตรมาส", |
| "question": "เรากำลังไปทางไหน ควรตั้งเป้าเท่าไหร่ และต้องเตรียมเงินสดแค่ไหน", |
| }, |
| "operational": { |
| "th": "เชิงปฏิบัติการ", "en": "Operational", |
| "user": "หัวหน้าฝ่ายขาย / หัวหน้าทีม", |
| "sees": "สถานะล่าสุดเทียบกับที่ควรจะเป็น", |
| "cadence": "ทบทวนรอบวัน – สัปดาห์", |
| "question": "เดือนนี้ผิดปกติไหม ต้องเข้าไปแก้อะไรตอนนี้", |
| }, |
| "analytical": { |
| "th": "เชิงวิเคราะห์", "en": "Analytical", |
| "user": "นักวิเคราะห์ข้อมูล / ทีม Data", |
| "sees": "เบื้องหลังตัวเลข — โมเดล ความผิดพลาด และเพดานของข้อมูล", |
| "cadence": "ใช้ตามต้องการ ปรับ filter เองได้", |
| "question": "ตัวเลขนี้เชื่อได้แค่ไหน และจะทำให้ดีขึ้นได้อย่างไร", |
| }, |
| } |
|
|
|
|
| def _kpi(label, value, sub, tone="neutral", explain="", unit=""): |
| return {"label": label, "value": value, "unit": unit, "sub": sub, |
| "tone": tone, "explain": explain} |
|
|
|
|
| def _forecast_core(rows, s, periods, level): |
| """คำนวณพยากรณ์แบบเดียวกับ /api/forecast — ใช้ร่วมกันในแดชบอร์ด""" |
| bt, bt_meta = _cached_backtest(rows, s, min(BACKTEST_HORIZON, max(1, periods))) |
| leaderboard, bench = build_leaderboard(bt) if bt else ([], None) |
| used = leaderboard[0]["model"] if leaderboard else ( |
| "holtwinters" if len(s) >= SEASONAL_MIN else "linear") |
| try: |
| mean = MODELS[used](s, periods) |
| if not np.all(np.isfinite(mean)): |
| raise ValueError |
| except Exception: |
| used = "seasonal_naive" |
| mean = MODELS[used](s, periods) |
| errs = bt.get(used, {}).get("errors_by_h", {}) if bt else {} |
| resid_std = float(np.std(np.diff(s.values))) if len(s) > 1 else float(np.std(s.values)) |
| lower, upper, ci = empirical_intervals(mean, errs, level, resid_std) |
| return {"bt": bt, "bt_meta": bt_meta, "leaderboard": leaderboard, "bench": bench, |
| "used": used, "mean": mean, "lower": lower, "upper": upper, "ci": ci} |
|
|
|
|
| def _coverage_from_bt(bt, model, level): |
| """ตรวจว่าช่วงความเชื่อมั่นครอบคลุมความจริงกี่ % — คำนวณจากความผิดพลาดใน backtest""" |
| if not bt or model not in bt: |
| return None |
| z = Z.get(int(level), Z[80]) |
| hit = tot = 0 |
| run = 0.0 |
| for h in sorted(bt[model]["errors_by_h"]): |
| e = np.asarray(bt[model]["errors_by_h"][h], dtype=float) |
| e = e[np.isfinite(e)] |
| if len(e) < 3: |
| continue |
| run = max(run, float(np.std(e, ddof=1))) |
| hit += int(np.sum(np.abs(e) <= z * run)) |
| tot += len(e) |
| return None if not tot else round(100 * hit / tot, 1) |
|
|
|
|
| @app.route("/api/dashboard", methods=["POST", "OPTIONS"]) |
| def dashboard(): |
| """แดชบอร์ดตามบทบาท — ข้อมูลชุดเดียวกัน แต่คนละมุมมองและคนละข้อเสนอ |
| |
| body: {data, role: "strategic"|"operational"|"analytical", periods, interval, drop_last_incomplete} |
| """ |
| if request.method == "OPTIONS": |
| return ("", 204) |
| try: |
| body = request.get_json(force=True) or {} |
| rows = body.get("data", []) |
| role = str(body.get("role", "strategic")).lower() |
| if role not in ROLES: |
| return jsonify({"error": f"role ต้องเป็นหนึ่งใน {list(ROLES)}"}), 400 |
| periods = max(1, min(24, int(body.get("periods", 6)))) |
| level = int(body.get("interval", 80)) |
| level = level if level in Z else 80 |
| drop_last = bool(body.get("drop_last_incomplete", True)) |
|
|
| if len(rows) < MIN_HISTORY: |
| return jsonify({"error": f"ต้องการข้อมูลอย่างน้อย {MIN_HISTORY} เดือน"}), 400 |
|
|
| s, warns = to_series(rows) |
| used_rows = rows |
| if drop_last and len(s) > MIN_HISTORY: |
| warns.append(f"ตัดเดือนสุดท้าย ({s.index[-1].strftime('%Y-%m')}) ออกก่อนคำนวณ") |
| s = s.iloc[:-1] |
| used_rows = rows[:-1] |
|
|
| q = data_quality(s) |
| v = np.asarray(s.values, dtype=float) |
| n = len(v) |
| fc = _forecast_core(used_rows, s, periods, level) |
| acc = fc["leaderboard"][0] if fc["leaderboard"] else None |
|
|
| kpis, charts, alerts, solutions, limits = [], [], [], [], [] |
|
|
| |
| if role == "strategic": |
| last12 = float(np.sum(v[-12:])) if n >= 12 else float(np.sum(v)) |
| prev12 = float(np.sum(v[-24:-12])) if n >= 24 else None |
| yoy = None if not prev12 else (last12 - prev12) / prev12 * 100 |
| tot_fc = float(np.sum(fc["mean"])) |
| sd_tot = float(fc["ci"]["sigma_by_horizon"][-1]) * np.sqrt(periods) |
| worst = max(0.0, tot_fc - Z[level] * sd_tot) |
| best = tot_fc + Z[level] * sd_tot |
| agg_mape = (round(fc["bt"][fc["used"]]["metrics_aggregate"]["mape"], 1) |
| if fc["bt"] and fc["used"] in fc["bt"] else None) |
|
|
| kpis = [ |
| _kpi("รายได้ 12 เดือนล่าสุด", round(last12, 2), |
| "เทียบ 12 เดือนก่อนหน้า " + ("ไม่มีข้อมูลพอ" if yoy is None else f"{yoy:+.1f}%"), |
| "good" if (yoy or 0) >= 0 else "bad", unit="บาท", |
| explain="ยอดรวมจริงที่เก็บได้ ไม่ใช่ค่าพยากรณ์"), |
| _kpi(f"คาดการณ์ {periods} เดือนข้างหน้า", round(tot_fc, 2), |
| f"ช่วงที่เป็นไปได้ {worst:,.0f} – {best:,.0f} บาท", "neutral", unit="บาท", |
| explain="ตัวเลขกลาง ไม่ใช่คำสัญญา — ให้ดูช่วงประกอบเสมอ"), |
| _kpi("กรณีแย่ที่ควรวางแผนรองรับ", round(worst, 2), |
| f"ขอบล่างของช่วง {level}%", "warn", unit="บาท", |
| explain="ใช้ตัวเลขนี้วางแผนกระแสเงินสด ไม่ใช่ตัวเลขกลาง"), |
| _kpi("ความคลาดเคลื่อนของยอดรวม", |
| "ไม่ทราบ" if agg_mape is None else agg_mape, |
| "วัดจากการทดสอบย้อนหลังจริง", "neutral", unit="%", |
| explain="ยอดรวมทั้งช่วงแม่นกว่ารายเดือนเสมอ เพราะความผิดพลาดหักล้างกัน"), |
| ] |
| charts = [ |
| {"block": "yoy_annual", "title": "ยอดรวมรายปีและอัตราเติบโต", |
| "why": "เห็นทิศทางระยะยาวในภาพเดียว — ระวังปีที่ยังไม่ครบ 12 เดือน"}, |
| {"block": "ytd", "title": "จังหวะสะสมของปีนี้เทียบปีก่อน", |
| "why": "ตอบคำถาม \"ปีนี้เร็วกว่าหรือช้ากว่าปีที่แล้ว\" ได้ตรงที่สุด"}, |
| {"block": "moving_average", "title": "แนวโน้มระยะยาว (เส้นเฉลี่ย 12 เดือน)", |
| "why": "ตัดความแกว่งรายเดือนออก เหลือเฉพาะทิศทาง"}, |
| ] |
| if yoy is not None and yoy < 0: |
| alerts.append({"level": "warn", |
| "text": f"รายได้ 12 เดือนล่าสุดลดลง {abs(yoy):.1f}% จากช่วงเดียวกันปีก่อน", |
| "action": "เปิดมุมมองเชิงปฏิบัติการเพื่อดูว่าเดือนไหนเป็นตัวฉุด"}) |
| if agg_mape is not None and agg_mape > 20: |
| alerts.append({"level": "warn", |
| "text": f"ความคลาดเคลื่อนของยอดรวมอยู่ที่ {agg_mape}% — สูงเกินกว่าจะผูก KPI แบบตายตัว", |
| "action": "ตั้งเป้าเป็นช่วง ไม่ใช่ตัวเลขเดียว"}) |
| solutions = [ |
| {"decision": "ตั้งเป้ายอดขาย", |
| "do": f"ตั้งเป้าเป็นยอดรวม {periods} เดือน ไม่ใช่รายเดือน", |
| "because": "พยากรณ์ยอดรวมคลาดเคลื่อนน้อยกว่ารายเดือนอย่างมีนัยสำคัญ", |
| "owner": "ฝ่ายบริหาร + ฝ่ายขาย"}, |
| {"decision": "วางแผนกระแสเงินสด", |
| "do": f"ใช้ตัวเลขกรณีแย่ {worst:,.0f} บาท เป็นฐานการวางแผน", |
| "because": "ถ้าวางแผนด้วยค่ากลาง มีโอกาสราวครึ่งหนึ่งที่เงินจะไม่พอ", |
| "owner": "CFO"}, |
| {"decision": "รอบการทบทวน", |
| "do": "ทบทวนตัวเลขทุกเดือน และปรับเป้าทุกไตรมาส", |
| "because": "พยากรณ์คือการประมาณ ณ เวลาหนึ่ง ไม่ใช่คำสัญญาที่ตายตัว", |
| "owner": "ฝ่ายบริหาร"}, |
| {"decision": "ลงทุนเพื่อให้พยากรณ์แม่นขึ้น", |
| "do": "สั่งให้เริ่มเก็บ pipeline รายดีลพร้อมวันคาดปิดและมูลค่า", |
| "because": "ความคลาดเคลื่อนที่เหลือมาจากข้อมูลที่ยังไม่ได้เก็บ ไม่ใช่จากโมเดล", |
| "owner": "ฝ่ายบริหาร + IT"}, |
| ] |
| limits = ["บอกไม่ได้ว่าเดือนไหนจะสูงหรือต่ำ — ใช้ดูภาพรวมเท่านั้น", |
| "ไม่รวมผลของดีลก้อนใหญ่ที่ยังไม่เกิด", |
| "ไม่ใช่เครื่องมือติดตามงานรายวัน ให้ใช้มุมมองเชิงปฏิบัติการแทน"] |
|
|
| |
| elif role == "operational": |
| last = float(v[-1]) |
| ly = float(v[-13]) if n >= 13 else None |
| avg3 = float(np.mean(v[-4:-1])) if n >= 4 else None |
| vs_ly = None if not ly else (last - ly) / ly * 100 |
| vs_avg3 = None if not avg3 else (last - avg3) / avg3 * 100 |
| below = int(np.sum(v[-12:] < np.mean(v[-12:]))) if n >= 12 else 0 |
| nxt = float(fc["mean"][0]) |
| nxt_lo, nxt_hi = float(fc["lower"][0]), float(fc["upper"][0]) |
|
|
| kpis = [ |
| _kpi(f"ยอดเดือนล่าสุด ({s.index[-1].strftime('%Y-%m')})", round(last, 2), |
| "เทียบเดือนเดียวกันปีก่อน " + ("ไม่มีข้อมูล" if vs_ly is None else f"{vs_ly:+.1f}%"), |
| "good" if (vs_ly or 0) >= 0 else "bad", unit="บาท", |
| explain="เทียบกับเดือนเดียวกันปีก่อน ไม่ใช่เดือนก่อนหน้า เพราะตัดผลฤดูกาลออก"), |
| _kpi("เทียบค่าเฉลี่ย 3 เดือนก่อนหน้า", |
| "ไม่มีข้อมูล" if vs_avg3 is None else round(vs_avg3, 1), |
| "" if avg3 is None else f"ค่าเฉลี่ย 3 เดือนก่อน {avg3:,.0f} บาท", |
| "good" if (vs_avg3 or 0) >= 0 else "warn", unit="%", |
| explain="บอกว่าโมเมนตัมระยะสั้นกำลังขึ้นหรือลง"), |
| _kpi("คาดการณ์เดือนถัดไป", round(nxt, 2), |
| f"ช่วงที่เป็นไปได้ {nxt_lo:,.0f} – {nxt_hi:,.0f} บาท", "neutral", unit="บาท", |
| explain="ถ้ายอดจริงหลุดออกนอกช่วงนี้ ถือว่าผิดปกติ ต้องหาสาเหตุ"), |
| _kpi("เดือนที่ต่ำกว่าค่าเฉลี่ย (12 เดือนล่าสุด)", below, |
| f"จาก {min(12, n)} เดือน", "warn" if below >= 7 else "neutral", unit="เดือน", |
| explain="ถ้าเกินครึ่ง แปลว่ายอดกระจุกอยู่ที่ไม่กี่เดือนใหญ่"), |
| ] |
| charts = [ |
| {"block": "moving_average", "title": "ยอดรายเดือน 12–24 เดือนล่าสุด", |
| "why": "เห็นจังหวะขึ้นลงระยะสั้นที่ต้องเข้าไปจัดการ"}, |
| {"block": "growth", "title": "อัตราเติบโต MoM และ YoY", |
| "why": "แท่งติดลบติดกันหลายเดือน = สัญญาณที่ต้องรีบดู"}, |
| {"block": "month_profile", "title": "เดือนไหนแรงเดือนไหนอ่อนเป็นปกติ", |
| "why": "ป้องกันการตกใจกับเดือนที่ปกติก็ต่ำอยู่แล้ว"}, |
| ] |
| if n >= 13 and vs_ly is not None and vs_ly < -20: |
| alerts.append({"level": "critical", |
| "text": f"เดือนล่าสุดต่ำกว่าเดือนเดียวกันปีก่อน {abs(vs_ly):.0f}%", |
| "action": "ตรวจ pipeline ว่ามีดีลค้างที่ควรปิดแล้วยังไม่ปิดหรือไม่"}) |
| for o in q["outliers"][-2:]: |
| alerts.append({"level": "info", |
| "text": f"{o['period']} เป็นเดือนที่ยอดสูงผิดปกติ ({o['value']:,.0f} บาท)", |
| "action": "ตรวจว่าเป็นดีลก้อนใหญ่จริง แล้วอย่าใช้เดือนนั้นเป็นฐานตั้งเป้า"}) |
| if any("เดือนล่าสุด" in x for x in q["notes"]): |
| alerts.append({"level": "warn", |
| "text": "เดือนล่าสุดมียอดต่ำผิดปกติ — น่าจะยังบันทึกข้อมูลไม่ครบ", |
| "action": "ตรวจกับฝ่ายบัญชีว่าปิดยอดเดือนนั้นครบหรือยังก่อนสรุป"}) |
| solutions = [ |
| {"decision": "ติดตามรายเดือน", |
| "do": f"ตั้งเกณฑ์เตือนที่ขอบล่างของช่วง คือ {nxt_lo:,.0f} บาท", |
| "because": "ยอดที่ต่ำกว่านี้ถือว่าผิดปกติจริง ไม่ใช่แค่ความแกว่งตามธรรมชาติ", |
| "owner": "หัวหน้าฝ่ายขาย"}, |
| {"decision": "อ่านยอดที่ตกให้ถูก", |
| "do": "เทียบกับเดือนเดียวกันปีก่อนเสมอ ไม่ใช่เทียบเดือนก่อนหน้า", |
| "because": "การเทียบเดือนต่อเดือนทำให้ตกใจกับผลของฤดูกาลที่เป็นเรื่องปกติ", |
| "owner": "หัวหน้าทีม"}, |
| {"decision": "ตั้งเป้ารายทีม", |
| "do": "กระจายเป้าตามน้ำหนักเดือนในกราฟโปรไฟล์รายเดือน ไม่ใช่หาร 12 เท่ากัน", |
| "because": "แต่ละเดือนมีศักยภาพต่างกันโดยธรรมชาติของธุรกิจ", |
| "owner": "หัวหน้าฝ่ายขาย"}, |
| {"decision": "คุณภาพข้อมูล", |
| "do": "ปิดยอดให้ครบก่อนวันสรุปทุกเดือน", |
| "because": "เดือนที่บันทึกไม่ครบทำให้ทั้งรายงานและโมเดลอ่านผิด", |
| "owner": "ฝ่ายบัญชี"}, |
| ] |
| limits = ["ไม่ได้บอกว่าดีลไหนกำลังจะปิด — ต้องดูจากระบบ CRM", |
| "ไม่รู้เหตุผลเบื้องหลังตัวเลข บอกได้แค่ว่าผิดปกติหรือไม่", |
| "ไม่เหมาะกับการรายงานผู้บริหาร ให้ใช้มุมมองเชิงกลยุทธ์แทน"] |
|
|
| |
| else: |
| st_trend = st_seas = resid_share = None |
| try: |
| from statsmodels.tsa.seasonal import STL |
| r = STL(s, period=SEASONAL_PERIOD, robust=True).fit() |
|
|
| def _stg(c): |
| a, b = np.var(r.resid), np.var(r.resid + c) |
| return round(float(max(0.0, 1 - a / b)), 3) if b > 0 else 0.0 |
| st_trend, st_seas = _stg(r.trend), _stg(r.seasonal) |
| resid_share = round(float(np.var(r.resid) / np.var(v)), 3) if np.var(v) else None |
| except Exception: |
| pass |
|
|
| cov = _coverage_from_bt(fc["bt"], fc["used"], level) |
| beats = sum(1 for x in fc["leaderboard"] if x["kind"] == "model" and x["beats_baseline"]) |
|
|
| kpis = [ |
| _kpi("ความคลาดเคลื่อนรายเดือน", None if not acc else acc["mape"], |
| "" if not acc else f"โมเดล {acc['label']}", "neutral", unit="%", |
| explain="วัดด้วย rolling-origin backtest ทุกค่าเป็น out-of-sample"), |
| _kpi("skill เทียบ baseline", None if not acc else acc["skill_vs_baseline"], |
| f"โมเดลที่ชนะ baseline {beats} ตัว", |
| "good" if acc and (acc["skill_vs_baseline"] or 0) > 0 else "bad", |
| explain="เป็นบวก = คุ้มที่จะใช้ · เป็นลบ = แพ้การเดาแบบง่าย"), |
| _kpi("สัดส่วนที่อธิบายไม่ได้", None if resid_share is None else round(resid_share * 100, 1), |
| f"ความแรงฤดูกาล {st_seas} · แนวโน้ม {st_trend}", |
| "bad" if (resid_share or 0) > 0.5 else "neutral", unit="%", |
| explain="นี่คือเพดานของงาน — เปลี่ยนโมเดลไม่ช่วย ต้องหาข้อมูลใหม่"), |
| _kpi(f"ช่วง {level}% ครอบคลุมจริง", "ไม่ทราบ" if cov is None else cov, |
| f"ตั้งใจไว้ {level}%", |
| "good" if cov is not None and abs(cov - level) <= 10 else "warn", unit="%", |
| explain="ต่ำกว่าที่ตั้งใจ = ช่วงแคบเกินจริง = มั่นใจเกินเหตุ"), |
| ] |
| charts = [ |
| {"block": "leaderboard", "title": "ตารางอันดับโมเดลทั้งหมด", |
| "why": "เทียบทุกโมเดลในสนามเดียวกัน พร้อม skill score", |
| "endpoint": "POST /api/backtest"}, |
| {"block": "hindcast", "title": "เส้นทำนายย้อนหลังทับเส้นจริง", |
| "why": "เห็นกับตาว่าโมเดลพลาดช่วงไหน และพลาดไปทางไหน", |
| "endpoint": "POST /api/hindcast"}, |
| {"block": "decompose", "title": "แยก trend / seasonal / residual", |
| "why": "หาว่าเพดานของข้อมูลอยู่ตรงไหน", "endpoint": "POST /api/decompose"}, |
| {"block": "distribution", "title": "การกระจายของยอดรายเดือน", |
| "why": "ดูว่าข้อมูลเบ้หรือมีหางยาว ซึ่งกระทบการเลือกมาตรวัด"}, |
| ] |
| if acc and (acc["skill_vs_baseline"] or 0) <= 0: |
| alerts.append({"level": "critical", |
| "text": "โมเดลที่ดีที่สุดยังไม่ชนะ baseline", |
| "action": "อย่าเพิ่งนำไปใช้ — รายงานตามจริงว่ายังพยากรณ์ได้ไม่ดีกว่าการเดา"}) |
| if resid_share is not None and resid_share > 0.5: |
| alerts.append({"level": "warn", |
| "text": f"ความผันผวน {resid_share*100:.0f}% อธิบายไม่ได้ด้วยแนวโน้มหรือฤดูกาล", |
| "action": "หา feature ภายนอกเพิ่ม เช่น pipeline หรือฤดูกาลงบประมาณลูกค้า"}) |
| if cov is not None and cov < level - 10: |
| alerts.append({"level": "critical", |
| "text": f"ช่วง {level}% ครอบคลุมจริงแค่ {cov}% — ช่วงแคบเกินจริง", |
| "action": "ขยายช่วงหรือเพิ่มจุดทดสอบ ก่อนนำตัวเลขไปรายงาน"}) |
| solutions = [ |
| {"decision": "จะปรับปรุงโมเดลต่อไหม", |
| "do": ("ลองเพิ่มตัวแปรภายนอกก่อน แล้วค่อยลองโมเดลที่ซับซ้อนขึ้น" |
| if (resid_share or 0) > 0.5 else "ปรับพารามิเตอร์ของโมเดลที่ชนะอยู่"), |
| "because": "สัดส่วนที่อธิบายไม่ได้สูง แปลว่าเพดานอยู่ที่ข้อมูล ไม่ใช่ที่โมเดล", |
| "owner": "ทีม Data"}, |
| {"decision": "หน่วยเวลาที่ควรรายงาน", |
| "do": "รายงานเป็นยอดรวมไตรมาส และแนบช่วงความเชื่อมั่นทุกครั้ง", |
| "because": "ความผิดพลาดรายเดือนหักล้างกันเมื่อรวมยอด", |
| "owner": "ทีม Data"}, |
| {"decision": "ตรวจสอบก่อนส่งมอบ", |
| "do": "รัน coverage test ทุกครั้งที่เปลี่ยนโมเดลหรือข้อมูล", |
| "because": "ช่วงที่ครอบคลุมต่ำกว่าที่ตั้งใจ อันตรายกว่าการไม่มีช่วงเลย", |
| "owner": "ทีม Data"}, |
| {"decision": "สื่อสารกับผู้บริหาร", |
| "do": "ส่งมุมมองเชิงกลยุทธ์ให้ผู้บริหาร ไม่ใช่หน้านี้", |
| "because": "ตัวเลข MAPE และ skill score ไม่ใช่ภาษาที่ใช้ตัดสินใจเชิงธุรกิจ", |
| "owner": "ทีม Data"}, |
| ] |
| limits = ["ไม่ใช่หน้าจอสำหรับตัดสินใจธุรกิจโดยตรง", |
| "ตัวเลขทุกตัวขึ้นกับจำนวนจุดทดสอบ — เปลี่ยนจำนวนจุด ตัวเลขเปลี่ยน", |
| "ไม่ได้ตรวจว่าข้อมูลต้นทางถูกบันทึกถูกต้องหรือไม่"] |
|
|
| order = ["strategic", "operational", "analytical"] |
| return jsonify({ |
| "role": role, |
| "role_info": ROLES[role], |
| "kpis": kpis, |
| "charts": charts, |
| "alerts": alerts, |
| "solutions": solutions, |
| "limits": limits, |
| "other_roles": [{"role": r, "th": ROLES[r]["th"], "user": ROLES[r]["user"], |
| "question": ROLES[r]["question"]} for r in order if r != role], |
| "shared": { |
| "model_used": fc["used"], |
| "model_used_th": MODEL_INFO.get(fc["used"], {}).get("th", fc["used"]), |
| "periods": periods, |
| "interval_level": level, |
| "history_range": f"{q['start']} → {q['end']}", |
| "n_months": n, |
| "forecast": [ |
| {"period": p, "predicted": round(float(m), 2), |
| "lower": round(float(lo), 2), "upper": round(float(hi), 2)} |
| for p, m, lo, hi in zip(next_periods(s.index[-1], periods), |
| fc["mean"], fc["lower"], fc["upper"]) |
| ], |
| }, |
| "note": ("ทั้งสามมุมมองใช้ข้อมูลและโมเดลชุดเดียวกัน — ต่างกันที่ " |
| "เลือกแสดงตัวเลขคนละชุด และเสนอการตัดสินใจคนละแบบ ตามสิ่งที่ผู้ใช้บทบาทนั้นทำได้จริง"), |
| "warnings": warns + q["notes"], |
| }) |
| except Exception as e: |
| return jsonify({"error": f"{type(e).__name__}: {e}"}), 500 |
|
|
|
|
| @app.route("/api/sample.xlsx") |
| def sample_xlsx(): |
| """ดาวน์โหลดข้อมูลตัวอย่างเป็นไฟล์ Excel พร้อมพจนานุกรมคอลัมน์และวิธีใช้""" |
| try: |
| from openpyxl import Workbook |
| from openpyxl.styles import Alignment, Font, PatternFill |
| from openpyxl.utils import get_column_letter |
|
|
| path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "sample_ent.json") |
| with open(path, encoding="utf-8") as f: |
| src = json.load(f) |
| data, extra = src["data"], {e["date"]: e for e in src.get("extra", [])} |
|
|
| wb = Workbook() |
| HEAD = PatternFill("solid", fgColor="1E2761") |
| HF_ = Font(color="FFFFFF", bold=True, size=11) |
|
|
| def style_header(ws, ncol): |
| for c in range(1, ncol + 1): |
| cell = ws.cell(row=1, column=c) |
| cell.fill, cell.font = HEAD, HF_ |
| cell.alignment = Alignment(horizontal="center", vertical="center") |
| ws.freeze_panes = "A2" |
|
|
| |
| ws = wb.active |
| ws.title = "ข้อมูลรายเดือน" |
| ws.append(["date", "sales", "qt_count", "qt_value", "win_rate"]) |
| for r in data: |
| e = extra.get(r["date"], {}) |
| ws.append([r["date"], r["sales"], e.get("qt_count"), e.get("qt_value"), e.get("win_rate")]) |
| style_header(ws, 5) |
| for col, w in zip("ABCDE", [12, 16, 12, 16, 12]): |
| ws.column_dimensions[col].width = w |
| for row in ws.iter_rows(min_row=2, min_col=2, max_col=2): |
| for c in row: |
| c.number_format = "#,##0.00" |
| for row in ws.iter_rows(min_row=2, min_col=4, max_col=4): |
| for c in row: |
| c.number_format = "#,##0.00" |
| for row in ws.iter_rows(min_row=2, min_col=5, max_col=5): |
| for c in row: |
| c.number_format = "0.0%" |
|
|
| |
| ws2 = wb.create_sheet("พจนานุกรมคอลัมน์") |
| ws2.append(["คอลัมน์", "ชื่อไทย", "ชนิดข้อมูล", "ตัวอย่าง", "ความหมาย"]) |
| for row in [ |
| ["date", "เดือน", "ข้อความ YYYY-MM", data[0]["date"], "เดือนที่ข้อมูลเกิดขึ้น หนึ่งแถวคือหนึ่งเดือน"], |
| ["sales", "รายได้ปิดจริง (บาท)", "ตัวเลข", f"{data[0]['sales']:,.2f}", "ยอดรวมจากใบเสร็จทั้งเดือน — คอลัมน์หลักที่ใช้พยากรณ์"], |
| ["qt_count", "จำนวนใบเสนอราคา", "ตัวเลข", "38", "จำนวนใบเสนอราคาที่ออกในเดือนนั้น (ต้นกรวยการขาย)"], |
| ["qt_value", "มูลค่าใบเสนอราคา (บาท)", "ตัวเลข", "2,500,000.00", "มูลค่ารวมของใบเสนอราคาในเดือนนั้น"], |
| ["win_rate", "อัตราชนะดีล", "ตัวเลข 0–1", "0.65", "สัดส่วนใบเสนอราคาที่ปิดได้ในเดือนนั้น"], |
| ]: |
| ws2.append(row) |
| style_header(ws2, 5) |
| for col, w in zip("ABCDE", [16, 26, 20, 22, 62]): |
| ws2.column_dimensions[col].width = w |
| for r in ws2.iter_rows(min_row=2): |
| r[4].alignment = Alignment(wrap_text=True, vertical="top") |
|
|
| |
| ws3 = wb.create_sheet("วิธีใช้") |
| ws3.append(["หัวข้อ", "รายละเอียด"]) |
| for row in [ |
| ["ชุดข้อมูลนี้คืออะไร", "ยอดขายรายเดือนของบริษัทตัวอย่าง (เรียกว่า บริษัท ABC) จากข้อมูลจริงที่ปกปิดชื่อแล้ว"], |
| ["ช่วงเวลา", f"{data[0]['date']} ถึง {data[-1]['date']} รวม {len(data)} เดือน"], |
| ["หนึ่งแถวคืออะไร", "หนึ่งเดือน ไม่ใช่หนึ่งรายการขาย — รายการทั้งหมดในเดือนนั้นถูกบวกรวมแล้ว"], |
| ["ข้อควรระวังข้อแรก", "เดือนสุดท้ายมักยังบันทึกข้อมูลไม่ครบ ทำให้ดูเหมือนยอดตก ต้องตัดออกก่อนฝึกโมเดล"], |
| ["ข้อควรระวังข้อสอง", "ข้อมูลผันผวนสูง (CV ประมาณ 0.40) ความผิดพลาดระดับหลายสิบเปอร์เซ็นต์เป็นเรื่องปกติ"], |
| ["ขั้นที่ 1", "ทำความรู้จักข้อมูล — ตรวจว่าเดือนครบไหม มีค่าสุดโต่งไหม (POST /api/describe)"], |
| ["ขั้นที่ 2", "ดูภาพรวมด้วยกราฟ — ยอดรายปี เทียบเดือนข้ามปี ตารางความร้อน (POST /api/charts)"], |
| ["ขั้นที่ 3", "ตรวจว่าพยากรณ์ได้แค่ไหน — แนวโน้มและฤดูกาลมีจริงไหม (POST /api/decompose)"], |
| ["ขั้นที่ 4", "เทียบโมเดลกับวิธีเดาแบบง่าย — ดูคอลัมน์ skill ต้องเป็นบวก (POST /api/backtest)"], |
| ["ขั้นที่ 5", "ทำนายย้อนกลับดูของจริง — เส้นทำนายทับเส้นจริง (POST /api/hindcast)"], |
| ["ขั้นที่ 6", "พยากรณ์อนาคตและอ่านผลให้เป็น — ดูช่วงและเหตุผลประกอบ (POST /api/forecast)"], |
| ["แดชบอร์ดตามบทบาท", "POST /api/dashboard พร้อม role = strategic / operational / analytical"], |
| ["หมายเหตุความเป็นส่วนตัว", "ไฟล์นี้เป็นยอดรวมรายเดือน ไม่มีชื่อลูกค้า เลขผู้เสียภาษี หรือชื่อพนักงานขาย"], |
| ]: |
| ws3.append(row) |
| style_header(ws3, 2) |
| ws3.column_dimensions["A"].width = 30 |
| ws3.column_dimensions["B"].width = 100 |
| for r in ws3.iter_rows(min_row=2): |
| r[1].alignment = Alignment(wrap_text=True, vertical="top") |
|
|
| buf = io.BytesIO() |
| wb.save(buf) |
| buf.seek(0) |
| from flask import send_file |
| return send_file(buf, mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", |
| as_attachment=True, download_name="sample_ABC_90months.xlsx") |
| except ImportError: |
| return jsonify({"error": "ต้องติดตั้ง openpyxl ก่อน — เพิ่ม openpyxl ลงใน requirements.txt"}), 500 |
| except Exception as e: |
| return jsonify({"error": f"{type(e).__name__}: {e}"}), 500 |
|
|
|
|
| if __name__ == "__main__": |
| app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 7860))) |
|
|
