Build and Deploy a Worker Node With Alibaba Cloud

# -*- coding: utf-8 -*-
 
import time                    # Built-in library, used for periodic loops and waiting (sleep)
import math                    # Built-in library, used for logarithmic calculations
import threading               # Built-in library, uses threads for scheduled tasks
import requests                # HTTP request library, used to call OKX REST API
import pickle                  # Used for model serialization (save/load)
import json                    # Used to handle JSON data format
from flask import Flask, Response, request  # Flask: used to provide a web interface
import numpy as np             # Numerical computation library
import statsmodels.api as sm   # OLS regression model library (for least squares training)
 
# Global configuration (modifiable):
TOKEN = "SOL-USDT"             # Product code, e.g., "SOL-USDT" (trading pair), can be changed
API_BASE = "https://www.okx.com"  # Base address of OKX API
MODEL_PATH = "ols_model.pkl"   # Local path to save the trained model file
UPDATE_INTERVAL = 300          # Interval for scheduled updates (seconds), here 5 minutes = 300 seconds
DATA_POINTS = 180              # Number of past days of daily data to fetch
 
app = Flask(__name__)          # Create Flask application instance
 
def getData():
    """
    Fetch past DATA_POINTS days of 1-day candlestick data (including closing prices).
    Uses OKX `/api/v5/market/candles` endpoint (allows up to 300 entries at a time).
    Returns a list of tuples (timestamp, closing price) sorted in ascending order (UTC).
    """
    # Calculate limit
    limit = DATA_POINTS if DATA_POINTS <= 300 else 300
    # Request parameters
    params = {
        "instId": TOKEN,
        "bar": "1D",
        "limit": str(limit)
    }
    # Send GET request
    resp = requests.get(f"{API_BASE}/api/v5/market/candles", params=params, timeout=10)
    resp.raise_for_status()  # Raise exception on error
    resp_json = resp.json()
    if resp_json.get("code") != "0":
        raise RuntimeError(f"OKX returned error: {resp_json.get('msg')}")
    # data = [ [ts, o, h, l, c, ...], ... ], where ts is the open time in milliseconds
    data = resp_json["data"]
    # Convert to ascending order and extract [ts, closing price]
    result = sorted([(int(item[0]) // 1000, float(item[4])) for item in data], key=lambda x: x[0])
    return result
 
def compute_log_returns(series):
    """
    Accepts a list of tuples (ts, price) sorted in ascending order by ts.
    Calculates daily log returns ln(p_{t+1} / p_t).
    Returns a numpy array X (features) and a target vector y:
      X = [[price_t, ts_t], …] excluding the last day;
      y = [ln(price_{t+1}/price_t), …] corresponding to each X[i].
    """
    n = len(series)
    if n < 2:
        raise ValueError("Not enough data points to compute log-return")
    prices = np.array([p for (_, p) in series], dtype=float)
    timestamps = np.array([t for (t, _) in series], dtype=float)
    log_returns = np.log(prices[1:] / prices[:-1])
    # Features can include only price, or also timestamp to capture trends
    X = np.column_stack((prices[:-1], timestamps[:-1]))
    y = log_returns
    return X, y
 
def update_loop():
    """
    Background thread function: every UPDATE_INTERVAL seconds,
    automatically calls update_task().
    """
    while True:
        try:
            update_task()
        except Exception as e:
            # Do not stop thread on error, just print it
            print("update_task error:", e)
        time.sleep(UPDATE_INTERVAL)
 
def update_task():
    """
    Main update logic:
    1. Fetch past DATA_POINTS daily data;
    2. Compute log-returns;
    3. Fit an OLS least squares model;
    4. Save the model to a local file.
    """
    series = getData()
    X, y = compute_log_returns(series)
    # Add intercept term
    X_with_const = sm.add_constant(X)
    model = sm.OLS(y, X_with_const).fit()
    with open(MODEL_PATH, "wb") as f:
        pickle.dump(model, f)
    # No return after training, model saved successfully
    print("Model trained and saved.")
 
@app.route("/inference/<string:token>")
def generate_inference(token):
    """
    Route `/inference/<token>`:
    Checks if the token matches the global TOKEN (case-insensitive).
    Then fetches the latest two days' prices using getData,
    uses the saved model to predict the next log-return,
    and computes the predicted price as c_t * exp(pred_return).
    Responds with JSON containing the predicted log-return and predicted price.
    """
    if token.upper() != TOKEN.upper():
        # Token mismatch, return 400
        return Response(json.dumps({"error": "Token not supported"}), status=400, mimetype="application/json")
    # Load the trained model
    try:
        with open(MODEL_PATH, "rb") as f:
            model = pickle.load(f)
    except FileNotFoundError:
        return Response(json.dumps({"error": "Model not trained yet"}), status=500, mimetype="application/json")
    # Get the latest two daily data points
    series = getData()
    if len(series) < 2:
        # Not enough data for prediction
        return Response(json.dumps({"error": "Not enough data for inference"}), status=500, mimetype="application/json")
    # Latest ts, price
    ts_current, price_current = series[-2]  # Second last entry is the "current" parameters
    ts_next, price_next = series[-1]       # Last entry is the "actual next day" price
    # Prepare feature vector
    X_pred = np.array([[price_current, ts_current]])
    X_pred_with_const = sm.add_constant(X_pred, has_constant="add")
    pred_log_return = float(model.predict(X_pred_with_const)[0])
 
    return Response(str(pred_log_return), status=200, mimetype="application/json")
 
@app.route("/update")
def http_update():
    """
    Another endpoint `/update`: manually/externally triggers an update task (train model) once.
    Returns "0" for success, "1" for failure.
    """
    try:
        update_task()
        return "0"
    except Exception as e:
        print("Manual update error:", e)
        return "1"
 
if __name__ == "__main__":
    # When the script is run directly, train the model once first
    update_task()
    # Then start a background thread to continuously update
    thread = threading.Thread(target=update_loop, daemon=True)
    thread.start()
    # Start Flask web service
    app.run(host="0.0.0.0", port=8000)