File size: 2,087 Bytes
b90f45c 6e79e1c b90f45c 6de6d01 b90f45c 6de6d01 b90f45c 6de6d01 6e79e1c 6de6d01 6e79e1c 6de6d01 6e79e1c 6de6d01 6e79e1c 6de6d01 6e79e1c 6de6d01 6e79e1c 6de6d01 6e79e1c 6de6d01 6e79e1c 6de6d01 6e79e1c 6de6d01 6e79e1c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | import requests
import time
import pandas as pd
import gradio as gr
from fastapi import FastAPI
import threading
app = FastAPI()
URL = "https://histdatafeed.vps.com.vn/tradingview/history"
# ===== CACHE GLOBAL =====
latest_data = None
latest_signal = None
last_update = 0
# ===== FETCH DATA =====
def fetch_data():
global latest_data, last_update
try:
params = {
"symbol": "VN30F1M",
"resolution": "1",
"from": int(time.time()) - 3600,
"to": int(time.time()),
"countback": 200
}
res = requests.get(URL, params=params, timeout=5)
data = res.json()
df = pd.DataFrame({
"time": data["t"],
"open": data["o"],
"high": data["h"],
"low": data["l"],
"close": data["c"],
"volume": data["v"]
})
latest_data = df
last_update = time.time()
except Exception as e:
print("Fetch error:", e)
# ===== AI SIGNAL =====
def compute_signal():
global latest_data, latest_signal
if latest_data is None or len(latest_data) < 20:
return "WAIT"
df = latest_data.copy()
df["ma20"] = df["close"].rolling(20).mean()
if df["close"].iloc[-1] > df["ma20"].iloc[-1]:
latest_signal = "BUY"
else:
latest_signal = "SELL"
# ===== HEARTBEAT LOOP =====
def heartbeat():
while True:
fetch_data()
compute_signal()
print("Heartbeat update:", latest_signal)
time.sleep(10) # 10s update
threading.Thread(target=heartbeat, daemon=True).start()
# ===== API =====
@app.get("/signal")
def signal():
return {
"signal": latest_signal,
"price": float(latest_data["close"].iloc[-1]) if latest_data is not None else 0,
"time": int(last_update)
}
# ===== UI =====
def ui():
return f"Signal: {latest_signal} | Updated: {int(last_update)}"
interface = gr.Interface(
fn=ui,
inputs=[],
outputs="text",
live=True
).queue()
app = gr.mount_gradio_app(app, interface, path="/") |