Create app.py

app.py

@@ -0,0 +1,178 @@
+import gradio as gr
+from vnstock import Vnstock
+from datetime import datetime, timedelta
+import pandas as pd
+import numpy as np
+from sklearn.linear_model import LinearRegression
+from sklearn.preprocessing import MinMaxScaler
+import torch
+from torch_geometric.data import Data
+
+vn = Vnstock()
+
+# ============================
+# TECHNICAL INDICATORS
+# ============================
+def calc_RSI(series, period=14):
+    delta = series.diff()
+    gain = delta.clip(lower=0)
+    loss = -delta.clip(upper=0)
+
+    avg_gain = gain.rolling(period).mean()
+    avg_loss = loss.rolling(period).mean()
+
+    RS = avg_gain / avg_loss
+    RSI = 100 - (100 / (1 + RS))
+    return RSI
+
+def calc_MACD(series, fast=12, slow=26, signal=9):
+    ema_fast = series.ewm(span=fast, adjust=False).mean()
+    ema_slow = series.ewm(span=slow, adjust=False).mean()
+    macd = ema_fast - ema_slow
+    signal_line = macd.ewm(span=signal, adjust=False).mean()
+    hist = macd - signal_line
+    return macd, signal_line, hist
+
+def calc_bollinger(series, window=20, num_std=2):
+    sma = series.rolling(window).mean()
+    std = series.rolling(window).std()
+    upper = sma + num_std * std
+    lower = sma - num_std * std
+    return sma, upper, lower
+
+# ============================
+# GNN MODEL
+# ============================
+class StockGCN(torch.nn.Module):
+    def __init__(self, num_features, hidden=16):
+        super().__init__()
+        from torch_geometric.nn import GCNConv
+        self.conv1 = GCNConv(num_features, hidden)
+        self.conv2 = GCNConv(hidden, 1)
+
+    def forward(self, data):
+        x, edge = data.x, data.edge_index
+        x = torch.relu(self.conv1(x, edge))
+        x = self.conv2(x, edge)
+        return x
+
+# ============================
+# API IMPLEMENTATION
+# ============================
+
+def api_history(symbol, start, end):
+    try:
+        stock = vn.stock(symbol=symbol)
+        df = stock.quote.history(start=start, end=end, interval="1D")
+
+        if "close" in df.columns:
+            df = df.rename(columns={"close": "Close"})
+        if "time" in df.columns:
+            df = df.rename(columns={"time": "Date"}).set_index("Date")
+
+        return df.to_dict()
+    except Exception as e:
+        return {"error": str(e)}
+
+def api_ta(symbol, start, end):
+    try:
+        stock = vn.stock(symbol=symbol)
+        df = stock.quote.history(start=start, end=end, interval="1D")
+
+        df["RSI"] = calc_RSI(df["close"])
+        df["MACD"], df["MACD_signal"], df["MACD_hist"] = calc_MACD(df["close"])
+        df["BB_MID"], df["BB_UPPER"], df["BB_LOWER"] = calc_bollinger(df["close"])
+
+        return df.fillna(None).to_dict()
+    except Exception as e:
+        return {"error": str(e)}
+
+def api_gnn(symbol, days):
+    try:
+        end = datetime.today()
+        start = end - timedelta(days=365)
+
+        stock = vn.stock(symbol=symbol)
+        df = stock.quote.history(start=start.strftime("%Y-%m-%d"),
+                                 end=end.strftime("%Y-%m-%d"),
+                                 interval="1D")
+        df = df.rename(columns={"close": "Close"})
+        df = df[["Close"]].dropna()
+
+        scaler = MinMaxScaler()
+        scaled = scaler.fit_transform(df.values)
+        df_scaled = pd.DataFrame(scaled, index=df.index, columns=["Close"])
+
+        # build graph chain
+        edge_index = torch.tensor([[i, i+1] for i in range(len(df_scaled)-1)],
+                                  dtype=torch.long).t().contiguous()
+
+        x = torch.tensor(df_scaled.values, dtype=torch.float)
+        data_obj = Data(x=x, edge_index=edge_index)
+
+        model = StockGCN(num_features=1)
+        model.eval()
+
+        preds_scaled = []
+        last_value = torch.tensor([[df_scaled.values[-1][0]]], dtype=torch.float)
+
+        for _ in range(days):
+            new_data = Data(
+                x=torch.cat([data_obj.x, last_value]),
+                edge_index=torch.tensor(
+                    [[i, i+1] for i in range(len(data_obj.x))],
+                    dtype=torch.long
+                ).t().contiguous()
+            )
+            out = model(new_data)
+            last_value = out[-1].view(1, 1)
+            preds_scaled.append(last_value.item())
+
+        preds_real = scaler.inverse_transform(
+            np.array(preds_scaled).reshape(-1, 1)
+        ).flatten()
+
+        dates = [(end + timedelta(days=i+1)).strftime("%Y-%m-%d") for i in range(days)]
+
+        return {
+            "symbol": symbol,
+            "today_close": float(df["Close"].iloc[-1]),
+            "predictions": [
+                {"date": d, "price": float(p)}
+                for d, p in zip(dates, preds_real)
+            ]
+        }
+    except Exception as e:
+        return {"error": str(e)}
+
+# ============================
+# GRADIO ROUTES
+# ============================
+
+with gr.Blocks() as app:
+    gr.Markdown("# 📈 VNStock API (Gradio on HuggingFace)")
+
+    with gr.Tab("History API"):
+        sym = gr.Text(label="Symbol")
+        start = gr.Text(label="Start")
+        end = gr.Text(label="End")
+        out = gr.JSON(label="Result")
+        btn = gr.Button("Get History")
+        btn.click(api_history, [sym, start, end], out)
+
+    with gr.Tab("Technical Analysis API"):
+        sym2 = gr.Text(label="Symbol")
+        start2 = gr.Text(label="Start")
+        end2 = gr.Text(label="End")
+        out2 = gr.JSON(label="Result")
+        btn2 = gr.Button("Get Indicators")
+        btn2.click(api_ta, [sym2, start2, end2], out2)
+
+    with gr.Tab("GNN Prediction API"):
+        sym3 = gr.Text(label="Symbol")
+        days3 = gr.Number(label="Days to Predict", value=7)
+        out3 = gr.JSON(label="Result")
+        btn3 = gr.Button("Predict GNN")
+        btn3.click(api_gnn, [sym3, days3], out3)
+
+app.launch()