Tardis API กับ Plotly Dash: สร้าง Data Visualization สำหรับ Dashboard วิเคราะห์ข้อมูลแบบ Interactive

ในยุคที่ข้อมูลคือสินทรัพย์สำคัญของธุรกิจ การสร้าง Dashboard ที่สามารถแสดงผลข้อมูลแบบ Interactive และอัปเดตแบบ Real-time เป็นสิ่งจำเป็นอย่างยิ่งสำหรับทีม Data Science และ Developer หลายคนอาจคุ้นเคยกับการใช้ Matplotlib หรือ Seaborn ในการสร้างกราฟแบบ Static แต่สำหรับ Dashboard ที่ต้องการความ Interactive สูง Plotly Dash คือตัวเลือกที่ยอดเยี่ยมที่สุดในปัจจุบัน ในบทความนี้เราจะมาเรียนรู้วิธีการใช้ Tardis API เพื่อดึงข้อมูลและสร้าง Interactive Dashboard ด้วย Plotly Dash กันอย่างละเอียด

Tardis API คืออะไร และทำไมต้องใช้กับ Data Visualization

Tardis API เป็น API ที่ให้บริการข้อมูล Real-time สำหรับการวิเคราะห์ในหลากหลายอุตสาหกรรม รวมถึงการวิเคราะห์พฤติกรรมลูกค้า E-commerce การติดตามตลาดการเงิน และการวิเคราะห์ระบบ AI/RAG เมื่อนำมารวมกับ Plotly Dash เราจะได้ Dashboard ที่สามารถ:

• แสดงผลข้อมูลแบบ Real-time Updates
• ให้ผู้ใช้สามารถ Filter และ Interact กับกราฟได้โดยตรง
• สร้าง Drill-down Analysis ที่ลึกและครอบคลุม
• Export ข้อมูลและกราฟในรูปแบบต่างๆ ได้อย่างง่ายดาย

การติดตั้ง Plotly Dash และการเตรียม Environment

ก่อนจะเริ่มสร้าง Dashboard เราต้องติดตั้ง Package ที่จำเป็นก่อน โดย Plotly Dash ต้องการ Python เวอร์ชัน 3.6 ขึ้นไป

# ติดตั้ง Plotly Dash และ Dependencies
pip install dash plotly pandas requests

หรือใช้ pipenv
pipenv install dash plotly pandas requests

ตรวจสอบเวอร์ชันที่ติดตั้ง
python -c "import dash; print(dash.__version__)"

โครงสร้างโปรเจกต์ที่แนะนำควรมีลักษณะดังนี้:

dashboard_project/
├── app.py              # ไฟล์หลักของ Dash Application
├── callbacks/
│   └── __init__.py
├── components/
│   └── __init__.py
├── data/
│   └── cache/          # เก็บข้อมูล Cache ชั่วคราว
├── pages/
│   └── analytics.py   # หน้าสำหรับ Analytics
├── services/
│   └── tardis_api.py  # Service สำหรับเรียก Tardis API
├── utils/
│   └── helpers.py
└── requirements.txt

การเชื่อมต่อ Tardis API เพื่อดึงข้อมูล

ในการใช้งาน Tardis API อย่างมีประสิทธิภาพ เราควรสร้าง Service Layer ที่จัดการเรื่อง Authentication, Error Handling และ Caching อย่างเป็นระบบ ต่อไ�นี้คือตัวอย่างการสร้าง Tardis API Service ที่ใช้งานได้จริง:

import requests
from typing import Dict, List, Optional, Any
from datetime import datetime, timedelta
import pandas as pd
import json
import hashlib

class TardisAPIService:
    """
    Service สำหรับเชื่อมต่อกับ Tardis API
    รองรับการดึงข้อมูล Real-time และ Historical Data
    """
    
    BASE_URL = "https://api.tardis.io/v1"  # ตัวอย่าง Tardis API
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
        self._cache: Dict[str, Any] = {}
        self._cache_ttl = 300  # Cache TTL ในวินาที
    
    def _generate_cache_key(self, endpoint: str, params: Dict) -> str:
        """สร้าง Cache Key จาก Endpoint และ Parameters"""
        key_string = f"{endpoint}:{json.dumps(params, sort_keys=True)}"
        return hashlib.md5(key_string.encode()).hexdigest()
    
    def _is_cache_valid(self, cache_key: str) -> bool:
        """ตรวจสอบว่า Cache ยัง valid หรือไม่"""
        if cache_key not in self._cache:
            return False
        cached_time = self._cache[cache_key].get("timestamp")
        if cached_time is None:
            return False
        elapsed = (datetime.now() - cached_time).total_seconds()
        return elapsed < self._cache_ttl
    
    def get_realtime_metrics(self, metric_type: str = "all") -> Dict:
        """
        ดึงข้อมูล Metrics แบบ Real-time
        
        Args:
            metric_type: ประเภทของ metric (all, sales, users, performance)
        
        Returns:
            Dictionary ที่มีข้อมูล Metrics
        """
        cache_key = self._generate_cache_key("realtime_metrics", {"type": metric_type})
        
        if self._is_cache_valid(cache_key):
            return self._cache[cache_key]["data"]
        
        try:
            response = self.session.get(
                f"{self.BASE_URL}/metrics/realtime",
                params={"type": metric_type},
                timeout=10
            )
            response.raise_for_status()
            data = response.json()
            
            # เก็บใน Cache
            self._cache[cache_key] = {
                "data": data,
                "timestamp": datetime.now()
            }
            return data
            
        except requests.exceptions.Timeout:
            # Fallback ไปยัง Cache เก่า หาก Timeout
            if cache_key in self._cache:
                return self._cache[cache_key]["data"]
            raise ConnectionError("Tardis API Timeout - กรุณาลองใหม่อีกครั้ง")
            
        except requests.exceptions.RequestException as e:
            raise ConnectionError(f"Tardis API Error: {str(e)}")
    
    def get_historical_data(
        self,
        start_date: datetime,
        end_date: datetime,
        data_type: str = "transactions"
    ) -> pd.DataFrame:
        """
        ดึงข้อมูล Historical Data ในรูปแบบ DataFrame
        
        Args:
            start_date: วันที่เริ่มต้น
            end_date: วันที่สิ้นสุด
            data_type: ประเภทข้อมูล (transactions, users, products)
        
        Returns:
            Pandas DataFrame ที่มีข้อมูล Historical
        """
        try:
            response = self.session.get(
                f"{self.BASE_URL}/data/historical",
                params={
                    "start": start_date.isoformat(),
                    "end": end_date.isoformat(),
                    "type": data_type
                },
                timeout=30
            )
            response.raise_for_status()
            data = response.json()
            
            # แปลงเป็น DataFrame
            df = pd.DataFrame(data.get("records", []))
            
            if not df.empty and "timestamp" in df.columns:
                df["timestamp"] = pd.to_datetime(df["timestamp"])
                df = df.sort_values("timestamp")
            
            return df
            
        except requests.exceptions.RequestException as e:
            raise ConnectionError(f"Failed to fetch historical data: {str(e)}")
    
    def get_ai_analytics(self, use_cache: bool = True) -> Dict:
        """
        ดึงข้อมูล Analytics จากระบบ AI โดยเชื่อมต่อผ่าน HolySheep API
        
        ตัวอย่างการใช้ HolySheep AI สำหรับ AI Analytics:
        - ราคาถูกกว่า 85%+ เมื่อเทียบกับ OpenAI
        - Latency ต่ำกว่า 50ms
        - รองรับ WeChat/Alipay
        """
        cache_key = "ai_analytics"
        
        if use_cache and self._is_cache_valid(cache_key):
            return self._cache[cache_key]["data"]
        
        try:
            # ใช้ HolySheep AI สำหรับ AI Analytics
            holysheep_response = self._call_holysheep_ai(
                prompt="วิเคราะห์ข้อมูลลูกค้าต่อไปนี้และสรุป Insights"
            )
            
            self._cache[cache_key] = {
                "data": holysheep_response,
                "timestamp": datetime.now()
            }
            return holysheep_response
            
        except Exception as e:
            raise RuntimeError(f"AI Analytics Error: {str(e)}")
    
    def _call_holysheep_ai(self, prompt: str) -> Dict:
        """เรียก HolySheep AI API สำหรับการวิเคราะห์"""
        holysheep_url = "https://api.holysheep.ai/v1/chat/completions"
        
        payload = {
            "model": "gpt-4.1",
            "messages": [
                {"role": "system", "content": "คุณคือ AI Analyst ที่ช่วยวิเคราะห์ข้อมูล"},
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.7
        }
        
        response = self.session.post(
            holysheep_url,
            headers={
                "Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
                "Content-Type": "application/json"
            },
            json=payload,
            timeout=15
        )
        response.raise_for_status()
        return response.json()


ตัวอย่างการใช้งาน
tardis_service = TardisAPIService(api_key="your_tardis_api_key")
metrics = tardis_service.get_realtime_metrics(metric_type="sales")

การสร้าง Plotly Dash Dashboard - โครงสร้างพื้นฐาน

ต่อไ�จะเป็นตัวอย่างการสร้าง Dash Application ที่มีโครงสร้างครบถ้วน รองรับ Callback หลายตัว และสามารถเรียกใช้งาน Tardis API ได้อย่างมีประสิทธิภาพ:

from dash import Dash, html, dcc, callback, ctx, Input, Output, State, ALL
import plotly.express as px
import plotly.graph_objects as go
from dash.exceptions import PreventUpdate
import pandas as pd
from datetime import datetime, timedelta
from services.tardis_api import TardisAPIService

สร้าง Dash Application
app = Dash(__name__, title="Tardis Analytics Dashboard")
app.suppress_callback_exceptions = True

Initialize Service
tardis_service = TardisAPIService(api_key="your_tardis_api_key")

สไตล์ CSS
external_stylesheets = [
    "https://fonts.googleapis.com/css2?family=Kanit:wght@300;400;500;600;700&display=swap",
    "https://cdn.jsdelivr.net/npm/[email protected]/dist/css/bootstrap.min.css"
]

app.layout = html.Div([
    # Header Section
    html.Header([
        html.Div([
            html.H1("📊 Tardis Analytics Dashboard", className="dashboard-title"),
            html.P("Real-time Data Visualization powered by Plotly Dash", 
                   className="dashboard-subtitle")
        ], className="header-content")
    ], className="dashboard-header"),
    
    # Controls Section
    html.Div([
        html.Div([
            html.Label("เลือกช่วงเวลา:", className="control-label"),
            dcc.Dropdown(
                id="time-range-selector",
                options=[
                    {"label": "วันนี้", "value": "today"},
                    {"label": "7 วันล่าสุด", "value": "7days"},
                    {"label": "30 วันล่าสุด", "value": "30days"},
                    {"label": "90 วันล่าสุด", "value": "90days"},
                    {"label": "กำหนดเอง", "value": "custom"}
                ],
                value="7days",
                className="dropdown-control"
            )
        ], className="control-group"),
        
        html.Div([
            html.Label("ประเภทข้อมูล:", className="control-label"),
            dcc.Dropdown(
                id="data-type-selector",
                options=[
                    {"label": "ยอดขาย", "value": "sales"},
                    {"label": "ผู้ใช้งาน", "value": "users"},
                    {"label": "ประสิทธิภาพ", "value": "performance"},
                    {"label": "AI Analytics", "value": "ai"}
                ],
                value="sales",
                className="dropdown-control"
            )
        ], className="control-group"),
        
        # Refresh Button
        html.Button(
            [html.Span("🔄", className="btn-icon"), " อัปเดตข้อมูล"],
            id="refresh-button",
            n_clicks=0,
            className="btn btn-primary refresh-btn"
        ),
        
        # Auto-refresh Interval
        dcc.Interval(
            id="auto-refresh",
            interval=60 * 1000,  # อัปเดตทุก 60 วินาที
            n_intervals=0
        )
    ], className="controls-section"),
    
    # Metrics Cards
    html.Div([
        html.Div([
            html.Div([
                html.Span("💰", className="metric-icon"),
                html.Div([
                    html.H3(id="total-sales", className="metric-value"),
                    html.P("ยอดขายรวม", className="metric-label")
                ], className="metric-content")
            ], className="metric-card metric-sales")
        ], className="col-md-3"),
        
        html.Div([
            html.Div([
                html.Span("👥", className="metric-icon"),
                html.Div([
                    html.H3(id="total-users", className="metric-value"),
                    html.P("ผู้ใช้งาน", className="metric-label")
                ], className="metric-content")
            ], className="metric-card metric-users")
        ], className="col-md-3"),
        
        html.Div([
            html.Div([
                html.Span("📈", className="metric-icon"),
                html.Div([
                    html.H3(id="conversion-rate", className="metric-value"),
                    html.P("อัตราการแปลง", className="metric-label")
                ], className="metric-content")
            ], className="metric-card metric-conversion")
        ], className="col-md-3"),
        
        html.Div([
            html.Div([
                html.Span("⚡", className="metric-icon"),
                html.Div([
                    html.H3(id="avg-response-time", className="metric-value"),
                    html.P("เวลาตอบสนองเฉลี่ย", className="metric-label")
                ], className="metric-content")
            ], className="metric-card metric-performance")
        ], className="col-md-3")
    ], className="metrics-row"),
    
    # Charts Section
    html.Div([
        # Main Chart - Line Chart
        html.Div([
            html.H2("📈 แนวโน้มข้อมูล", className="chart-title"),
            dcc.Graph(id="main-line-chart", className="main-chart")
        ], className="chart-container col-lg-8"),
        
        # Pie Chart
        html.Div([
            html.H2("🥧 สัดส่วนประเภทสินค้า", className="chart-title"),
            dcc.Graph(id="pie-chart", className="pie-chart")
        ], className="chart-container col-lg-4"),
        
        # Bar Chart
        html.Div([
            html.H2("📊 เปรียบเทียบรายภูมิภาค", className="chart-title"),
            dcc.Graph(id="bar-chart", className="bar-chart")
        ], className="chart-container col-lg-6"),
        
        # Heatmap
        html.Div([
            html.H2("🔥 แผนที่ความร้อนกิจกรรม", className="chart-title"),
            dcc.Graph(id="heatmap-chart", className="heatmap-chart")
        ], className="chart-container col-lg-6")
    ], className="charts-section"),
    
    # Data Table
    html.Div([
        html.H2("📋 รายละเอียดข้อมูล", className="chart-title"),
        html.Div(id="data-table-container")
    ], className="table-container"),
    
    # Loading State
    dcc.Loading(
        id="loading-overlay",
        type="default",
        children=html.Div(id="loading-output"),
        fullscreen=True,
        overlay_style={"visibility": "visible", "opacity": 0.5}
    ),
    
    # Store for caching data
    dcc.Store(id="cached-data")
    
], className="dashboard-container")


==================== CALLBACKS ====================

@callback(
    Output("main-line-chart", "figure"),
    Output("pie-chart", "figure"),
    Output("bar-chart", "figure"),
    Output("heatmap-chart", "figure"),
    Output("total-sales", "children"),
    Output("total-users", "children"),
    Output("conversion-rate", "children"),
    Output("avg-response-time", "children"),
    Output("data-table-container", "children"),
    Output("cached-data", "data"),
    Input("time-range-selector", "value"),
    Input("data-type-selector", "value"),
    Input("refresh-button", "n_clicks"),
    Input("auto-refresh", "n_intervals")
)
def update_dashboard(time_range, data_type, n_clicks, n_intervals):
    """
    Callback หลักสำหรับอัปเดต Dashboard ทั้งหมด
    """
    try:
        # คำนวณช่วงเวลา
        end_date = datetime.now()
        if time_range == "today":
            start_date = end_date - timedelta(days=1)
        elif time_range == "7days":
            start_date = end_date - timedelta(days=7)
        elif time_range == "30days":
            start_date = end_date - timedelta(days=30)
        elif time_range == "90days":
            start_date = end_date - timedelta(days=90)
        else:
            start_date = end_date - timedelta(days=7)
        
        # ดึงข้อมูลจาก Tardis API
        df_historical = tardis_service.get_historical_data(
            start_date=start_date,
            end_date=end_date,
            data_type=data_type
        )
        
        # ดึง Metrics
        metrics = tardis_service.get_realtime_metrics(metric_type=data_type)
        
        # สร้าง Line Chart
        line_fig = create_line_chart(df_historical, data_type)
        
        # สร้าง Pie Chart
        pie_fig = create_pie_chart(df_historical)
        
        # สร้าง Bar Chart
        bar_fig = create_bar_chart(df_historical)
        
        # สร้าง Heatmap
        heatmap_fig = create_heatmap(df_historical)
        
        # สร้าง Data Table
        table = create_data_table(df_historical)
        
        return (
            line_fig,
            pie_fig,
            bar_fig,
            heatmap_fig,
            f"฿{metrics.get('total_sales', 0):,.0f}",
            f"{metrics.get('total_users', 0):,}",
            f"{metrics.get('conversion_rate', 0):.2f}%",
            f"{metrics.get('avg_response_time', 0):.0f}ms",
            table,
            df_historical.to_dict("records")
        )
        
    except Exception as e:
        # กรณีเกิด Error แสดง Empty State
        return create_empty_figures()


def create_line_chart(df: pd.DataFrame, data_type: str) -> go.Figure:
    """สร้าง Line Chart พร้อม Interactive Features"""
    if df.empty:
        return go.Figure()
    
    fig = go.Figure()
    
    colors = {
        "sales": "#4F46E5",
        "users": "#10B981", 
        "performance": "#F59E0B"
    }
    color = colors.get(data_type, "#4F46E5")
    
    fig.add_trace(go.Scatter(
        x=df["timestamp"] if "timestamp" in df.columns else df.index,
        y=df["value"] if "value" in df.columns else df.iloc[:, 0],
        mode="lines+markers",
        name=f"{data_type.upper()}",
        line=dict(color=color, width=3),
        marker=dict(size=8, symbol="circle"),
        hovertemplate="%{x}
" +
                      f"{data_type}: " + "%{y:,.2f}"
    ))
    
    # เพิ่ม Trend Line
    if len(df) > 1:
        x_numeric = range(len(df))
        y_values = df["value"] if "value" in df.columns else df.iloc[:, 0]
        
        fig.add_trace(go.Scatter(
            x=list(x_numeric),
            y=[y_values.iloc[0] + (y_values.iloc[-1] - y_values.iloc[0]) / len(df) * i 
               for i in x_numeric],
            mode="lines",
            name="Trend",
            line=dict(color="rgba(255,107,107,0.5)", width=2, dash="dash"),
            showlegend=True
        ))
    
    fig.update_layout(
        template="plotly_white",
        hovermode="x unified",
        legend=dict(orientation="h", yanchor="bottom", y=1.02),
        margin=dict(l=20, r=20, t=40, b=20),
        height=400
    )
    
    return fig


def create_pie_chart(df: pd.DataFrame) -> go.Figure:
    """สร้าง Pie Chart แสดงสัดส่วน"""
    if df.empty or "category" not in df.columns:
        return go.Figure()
    
    category_data = df.groupby("category")["value"].sum().reset_index()
    
    fig = go.Figure(data=[go.Pie(
        labels=category_data["category"],
        values=category_data["value"],
        hole=0.4,
        marker=dict(colors=px.colors.qualitative.Set3),
        textinfo="label+percent",
        textposition="outside"
    )])
    
    fig.update_layout(
        template="plotly_white",
        margin=dict(l=20, r=20, t=40, b=20),
        height=350,
        showlegend=True
    )
    
    return fig


def create_bar_chart(df: pd.DataFrame) -> go.Figure:
    """สร้าง Bar Chart เปรียบเทียบรายภูมิภาค"""
    if df.empty:
        return go.Figure()
    
    regions = ["กรุงเทพ", "ภาคกลาง", "ภาคเหนือ", "ภาคอีสาน", "ภาคใต้"]
    values = [df["value"].sum() * p for p in [0.35, 0.25, 0.15, 0.15, 0.10]]
    
    fig = go.Figure(data=[go.Bar(
        x=regions,
        y=values,
        marker_color=["#4F46E5", "#10B981", "#F59E0B", "#EF4444", "#8B5CF6"],
        text=[f"฿{v:,.0f}" for v in values],
        textposition="outside"
    )])
    
    fig.update_layout(
        template="plotly_white",
        margin=dict(l=20, r=20, t=40, b=20),
        height=350,
        xaxis_title="ภูมิภาค",
        yaxis_title="ยอดขาย (บาท)"
    )
    
    return fig


def create_heatmap(df: pd.DataFrame) -> go.Figure:
    """สร้าง Heatmap แสดงกิจกรรมตามช่วงเวลา"""
    hours = list(range(24))
    days = ["จันทร์", "อังคาร", "พุธ", "พฤหัสบดี", "ศุกร์", "เสาร์", "อาทิตย์"]
    
    # สร้างข้อมูลตัวอย่าง
    import numpy as np
    np.random.seed(42)
    data = np.random.randint(10, 100, size=(7, 24))
    
    fig = go.Figure(data=go.Heatmap(
        z=data,
        x=hours,
        y=days,
        colorscale="Viridis",
        hovertemplate="ชั่วโมง: %{x}:00
วัน: %{y}
กิจกรรม: %{z}"
    ))
    
    fig.update_layout(
        template="plotly_white",
        margin=dict(l=20, r=20, t=40, b=20),
        height=350,
        xaxis_title="ชั่วโมง",
        yaxis_title="วันในสัปดาห์"
    )
    
    return fig


def create_data_table(df: pd.DataFrame) -> html.Div:
    """สร้าง Data Table ที่สามารถ Sort ได้"""
    if df.empty:
        return html.Div("ไม่มีข้อมูล", className="empty-state")
    
    return html.Div([
        html.Table([
            html.Thead(
                html.Tr([
                    html.Th(col.replace("_", " ").title()) 
                    for col in df.columns[:6]  # แสดงเฉพาะ 6 คอลัมน์แรก
                ])
            ),
            html.Tbody([
                html.Tr([
                    html.Td(str(row[col])[:50] + "..." if len(str(row[col])) > 50 
                            else row[col])
                    for col in df.columns[:6]
                ])
                for _, row in df.head(10).iterrows()
            ])
        ], className="data-table")
    ], className="table-responsive")


def create_empty_figures() -> tuple:
    """สร้าง Empty Figures สำหรับกรณี Error"""
    empty_fig = go.Figure()
    empty_fig.update_layout(
        template="plotly_white",
        annotations=[dict(text="ไม่มีข้อมูล", x=0.5, y=0.5, 
                         showarrow=False, font=dict(size=20))]
    )
    return (empty_fig, empty_fig, empty_fig, empty_fig, 
            "฿0", "0", "0%", "0ms", html.Div("Error loading data"), {})


if __name__ == "__main__":
    app.run(debug=True, host="0.0.0.0", port=8050)

การปรับแต่ง CSS ให้ Dashboard สวยงาม

หากต้องการให้ Dashboard มีความสวยงามและเป็นมืออาชีพ สามาร�