目前的 ChainInsight 是一個 批次處理 (Batch Processing) 架構：
CSV 輸入 → ETL 清洗 → 統計分析 → 供應鏈優化 → ML 分析 → 23 張 PNG → 靜態 HTML 儀表板
每次都要手動跑 5 個 Python 腳本，圖表是靜態 PNG，儀表板只能讀已存在的圖和 CSV。
┌─────────────────────────────────────────────────┐
│ ChainInsight Live Platform │
└─────────────────────────────────────────────────┘
┌──────────────┐ ┌──────────────┐ ┌───────────────────────┐
│ Data Source │────▸│ Ingestion │────▸│ Event Bus / Queue │
│ │ │ Gateway │ │ (Redis Stream / │
│ • CSV Upload │ │ │ │ Celery / RQ) │
│ • API Push │ │ • Validate │ │ │
│ • ERP/WMS │ │ • Timestamp │ │ • on_data_arrived │
│ • Scheduled │ │ • Queue │ │ • on_etl_complete │
│ Polling │ │ │ │ • on_analysis_done │
└──────────────┘ └──────────────┘ └──────────┬────────────┘
│
┌──────────────────────────────────┐│
│ Pipeline Orchestrator ││
│ (自動觸發流程) │◂┘
└──────┬───────┬───────┬───────┬───┘
│ │ │ │
┌──────▾──┐ ┌──▾─────┐ ┌▾─────┐ ┌▾──────────┐
│ ETL │ │ Stats │ │ SCM │ │ ML + Planning │
│ Pipeline│ │Analysis│ │Optim.│ │ Engine │
│ │ │ │ │ │ │ │
│ 8-step │ │chart │ │EOQ │ │ 30 algos │
│ cleaning│ │01-08 │ │Monte │ │ + S&OP │
│ │ │Carlo │ │ │
└────┬────┘ └───┬────┘ └──┬───┘ └─────┬─────┘
│ │ │ │
▾ ▾ ▾ ▾
┌─────────────────────────────────────────┐
│ Result Store (DB/Cache) │
│ • SQLite / PostgreSQL (結構化資料) │
│ • Redis (即時 KPI 快取) │
│ • File Store (圖表 PNG/SVG) │
└──────────────────┬──────────────────────┘
│
┌──────────────────▾──────────────────────┐
│ Real-Time Dashboard (Web) │
│ │
│ ┌──────────┐ ┌──────────┐ ┌────────┐│
│ │WebSocket │ │ Plotly/ │ │ SSE ││
│ │ Server │ │ Echarts │ │ Push ││
│ └──────────┘ └──────────┘ └────────┘│
│ │
│ • KPI 即時更新 (無需刷新) │
│ • 互動式圖表 (zoom/filter/drill-down) │
│ • Pipeline 進度條 │
│ • S&OP 情境分析即時視覺化 │
└──────────────────────────────────────────┘
技術選型：FastAPI + Celery (或 RQ)
# app/main.py — FastAPI 入口
from fastapi import FastAPI, UploadFile, WebSocket
from app.pipeline import trigger_full_pipeline
from app.ws_manager import ConnectionManager
app = FastAPI(title="ChainInsight Live")
ws_manager = ConnectionManager()
@app.post("/api/ingest")
async def ingest_data(file: UploadFile):
"""新資料上傳後自動觸發完整 pipeline"""
raw_path = save_upload(file)
# 非同步觸發：ETL → Stats → SCM → ML → Capacity → Sensing → S&OP
task = trigger_full_pipeline.delay(raw_path)
return {"task_id": task.id, "status": "pipeline_started"}
@app.websocket("/ws/dashboard")
async def dashboard_ws(websocket: WebSocket):
"""WebSocket 即時推送分析結果到前端"""
await ws_manager.connect(websocket)
# 每個 pipeline 階段完成時推送更新Pipeline Orchestrator（自動化流程）：
# app/pipeline.py — 事件驅動的 Pipeline
from celery import chain
def trigger_full_pipeline(raw_path: str):
"""資料進來 → 自動跑完全部流程"""
workflow = chain(
etl_task.s(raw_path), # Step 1: 8-step ETL
stats_task.s(), # Step 2: 統計分析
supply_chain_task.s(), # Step 3: 供應鏈優化
ml_task.s(), # Step 4: ML 30 algos
sop_task.s(), # Step 5: S&OP 情境模擬
publish_results.s(), # Step 6: 推送到 Dashboard
)
return workflow.apply_async()技術選型比較：
| 方案 | 技術 | 優點 | 缺點 |
|---|---|---|---|
| A (推薦) | FastAPI + React/Vue + Plotly.js | 完全自定義、最佳 UX、WebSocket 原生 | 開發量較大 |
| B (快速) | Streamlit / Dash | Python 全棧、開發快 | 自定義受限、並行能力弱 |
推薦方案 A 的前端架構：
前端 (React + TypeScript)
├── components/
│ ├── KPIDashboard.tsx # 6 個 KPI 即時卡片
│ ├── PipelineProgress.tsx # 流程進度條 (ETL → Stats → ML → Capacity → Sensing → S&OP)
│ ├── InteractiveCharts/
│ │ ├── CorrelationMatrix.tsx # Plotly heatmap (可互動)
│ │ ├── MonteCarloSim.tsx # 動態 Monte Carlo 模擬
│ │ ├── EOQOptimizer.tsx # 拖動參數即時重算 EOQ
│ │ ├── SOPScenarioViz.tsx # S&OP 情境比較
│ │ └── InventoryHealthMap.tsx # ABC 分類 + 風險熱力圖
│ ├── DataUploader.tsx # 拖拽上傳 CSV
│ └── AlertPanel.tsx # 即時缺貨警報
├── hooks/
│ └── useWebSocket.ts # WebSocket 連線管理
└── stores/
└── dashboardStore.ts # 全域狀態管理
關鍵互動功能：
- 圖表全部從靜態 PNG → Plotly.js / ECharts 互動式（zoom、filter、drill-down）
- WebSocket 即時推送：pipeline 每階段完成 → 自動更新對應圖表
- 參數可調：使用者可以在前端調整 EOQ 的 S/H 參數、Monte Carlo 模擬次數、GA 的代數等
目前有 GA（遺傳演算法）做靜態安全庫存最佳化。S&OP 模擬的價值在於：它能透過情境分析比較不同策略的 KPI 表現，支持數據驅動的決策。
┌─────────────────────────────────────────────────────────┐
│ Demand Planning for Supply Chain │
├─────────────────────────────────────────────────────────┤
│ │
│ 模組 1: 產能規劃 (Capacity Planning) │
│ ───────────────────────────────────────── │
│ 輸入: 需求預測、產線 Profile、稼動率目標 │
│ 輸出: 可行性報告、瓶頸偵測、調整建議 │
│ │
│ 模組 2: 需求感知 (Demand Sensing) │
│ ───────────────────────────────────────── │
│ 輸入: POS 資料、社群訊號、天氣資料 │
│ 輸出: 調整後預測、需求突增偵測 │
│ │
│ 模組 3: S&OP 模擬器 (S&OP Simulator) │
│ ───────────────────────────────────────── │
│ 輸入: 基準預測、產能限制、供應限制 │
│ 輸出: 情境比較 (Fill Rate, 庫存成本, 稼動率) │
│ │
└─────────────────────────────────────────────────────────┘
從單一 CSV → 資料庫 + 版本化
# app/db/models.py
from sqlalchemy import Column, Integer, Float, String, DateTime, JSON
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class InventorySnapshot(Base):
"""每次資料匯入都保存一份快照，支援時間序列分析"""
__tablename__ = 'inventory_snapshots'
id = Column(Integer, primary_key=True)
ingested_at = Column(DateTime, index=True)
batch_id = Column(String, index=True)
product_id = Column(String, index=True)
category = Column(String)
unit_cost = Column(Float)
current_stock = Column(Float)
daily_demand_est = Column(Float)
safety_stock_target = Column(Float)
vendor_name = Column(String)
lead_time_days = Column(Float)
reorder_point = Column(Float)
stock_status = Column(String)
inventory_value = Column(Float)
class AnalysisResult(Base):
"""儲存每次分析結果，支援歷史比較"""
__tablename__ = 'analysis_results'
id = Column(Integer, primary_key=True)
batch_id = Column(String, index=True)
analysis_type = Column(String)
result_json = Column(JSON)
chart_path = Column(String)
created_at = Column(DateTime)
class SOPScenarioResult(Base):
"""儲存 S&OP 模擬情境結果"""
__tablename__ = 'sop_scenario_results'
id = Column(Integer, primary_key=True)
scenario_name = Column(String)
batch_id = Column(String, index=True)
fill_rate = Column(Float)
avg_utilization = Column(Float)
inventory_cost = Column(Float)
stockout_events = Column(Integer)
created_at = Column(DateTime)有了時間序列資料，就能解鎖之前標為 N/A 的演算法：
| 演算法 | 現狀 | 升級後 |
|---|---|---|
| RNN (#23) | N/A (缺時間序列) | 可用 — 多批次資料形成時間序列 |
| LSTM (#24) | N/A | 可用 — 需求預測 |
| Transformer (#25) | N/A | 可用 — 多品類注意力機制 |
| S&OP Simulation | N/A | 可用 — 情境比較 + KPI 分析 |
| Capacity Planning | N/A | 可用 — 瓶頸偵測 + 調整建議 |
| Demand Sensing | N/A | 可用 — 訊號處理 + 預測調整 |
升級後：30 個演算法全部 Applied！
# app/triggers.py — 多種資料輸入觸發方式
# 1. 檔案監控 (Watch 資料夾)
from watchdog.observers import Observer
class CSVWatcher:
"""監控指定資料夾，新 CSV 出現時自動觸發 pipeline"""
def on_created(self, event):
if event.src_path.endswith('.csv'):
trigger_full_pipeline.delay(event.src_path)
# 2. 排程觸發 (每日/每小時)
from celery.schedules import crontab
CELERYBEAT_SCHEDULE = {
'daily-inventory-sync': {
'task': 'app.pipeline.sync_from_erp',
'schedule': crontab(hour=6, minute=0),
},
'hourly-rl-update': {
'task': 'app.pipeline.sop_refresh',
'schedule': crontab(minute=0),
},
}
# 3. API Webhook (ERP/WMS 主動推送)
@app.post("/webhook/inventory-update")
async def erp_webhook(payload: InventoryUpdate):
trigger_full_pipeline.delay(payload.data)
# 4. 手動上傳 (Dashboard UI)
# → 見模組 1 的 /api/ingest 端點| 層級 | 技術 | 用途 |
|---|---|---|
| Web Framework | FastAPI | REST API + WebSocket |
| Task Queue | Celery + Redis | 非同步 pipeline 執行 |
| Database | SQLite (dev) → PostgreSQL (prod) | 結構化資料 + 時間序列 |
| Cache | Redis | 即時 KPI + Session |
| Planning | Capacity + Sensing + S&OP | 需求規劃引擎 |
| 前端 | React + Plotly.js (或 Streamlit 快速版) | 互動式儀表板 |
| 即時通訊 | WebSocket (FastAPI 內建) | 推送更新到前端 |
| 檔案監控 | Watchdog | 自動偵測新 CSV |
| 排程 | Celery Beat | 定時觸發 |
| 容器化 | Docker + docker-compose | 一鍵部署 |
① 將現有分析腳本模組化 (function-based → class-based)
② FastAPI 後端 + CSV 上傳 API
③ Pipeline orchestrator (ETL → Stats → ML 自動串接)
④ Streamlit 快速版即時儀表板 (取代靜態 HTML)
⑤ SQLite 儲存歷史分析結果
⑥ Capacity Planning (產能規劃、瓶頸偵測)
⑦ Demand Sensing (需求訊號處理、突增偵測)
⑧ S&OP Simulator (情境模擬、KPI 比較)
⑨ 產能 vs 需求視覺化 (utilization timeline, bottleneck chart)
⑩ S&OP 情境比較視覺化 (scenario comparison, demand-supply balance)
⑪ WebSocket 即時推送
⑫ React 互動式前端 (取代 Streamlit)
⑬ Watchdog 檔案監控自動觸發
⑭ Celery Beat 排程
⑮ 即時警報系統 (缺貨、異常)
⑯ PostgreSQL 遷移
⑰ Docker + docker-compose
⑱ LSTM/Transformer 需求預測 (需累積時間序列)
⑲ Multi-warehouse S&OP planning
⑳ CI/CD + 自動化測試
ChainInsight/
├── app/
│ ├── main.py # FastAPI 入口
│ ├── config.py # 統一設定
│ ├── pipeline/
│ │ ├── orchestrator.py # Pipeline 排程器
│ │ ├── etl.py # 重構的 ETL (from clean_data.py)
│ │ ├── stats.py # 重構的統計分析
│ │ ├── supply_chain.py # 重構的供應鏈優化
│ │ └── ml_engine.py # 重構的 ML 分析
│ ├── capacity/
│ │ ├── models.py # CapacityPlanner, bottleneck detection
│ │ └── visualization.py # Utilization timeline charts
│ ├── sensing/
│ │ ├── signals.py # SignalProcessor, spike detection
│ │ └── visualization.py # Signal timeline charts
│ ├── sop/
│ │ ├── simulator.py # SOPSimulator, scenario comparison
│ │ └── visualization.py # Demand-supply balance charts
│ ├── db/
│ │ ├── models.py # SQLAlchemy 模型
│ │ ├── session.py # DB 連線管理
│ │ └── migrations/ # Alembic 遷移
│ ├── api/
│ │ ├── routes.py # REST API 路由
│ │ └── websocket.py # WebSocket 管理
│ └── services/
│ ├── chart_service.py # 圖表產生 (Plotly)
│ ├── alert_service.py # 警報系統
│ └── export_service.py # PDF/Excel 匯出
├── frontend/ # React + TypeScript (Phase 3)
│ ├── src/
│ │ ├── components/
│ │ ├── hooks/
│ │ └── stores/
│ └── package.json
├── data/
│ ├── raw/ # 上傳的原始 CSV
│ ├── clean/ # ETL 清洗後
│ └── charts/ # 產生的圖表
├── tests/
├── docker-compose.yml
├── requirements.txt # 更新的依賴
└── README.md
| 面向 | 現在 | 升級後 |
|---|---|---|
| 觸發方式 | 手動跑 5 個腳本 | 上傳 CSV / API / 排程自動觸發 |
| 圖表 | 23 張靜態 PNG | 互動式 Plotly (zoom/filter/drill-down) |
| 儀表板 | 靜態 HTML 讀 CSV | WebSocket 即時串流更新 |
| 最佳化 | GA 一次性計算 | GA 全域搜索 + S&OP 情境模擬 |
| ML 覆蓋 | 20/30 演算法 | 30/30 全部 Applied |
| 資料儲存 | 單一 CSV | DB + 時間序列 + 歷史版本 |
| 部署 | python xxx.py |
Docker 一鍵啟動 |