DeworkHub｜基于 OLS 的 LAT 币价预测
大家都喜欢猜测未来的价格走势，今天 DeworkHub 带来基于 OLS 的 LAT 价格预测模型，废话不多说，直接上代码：
import json
from flask import Flask, Response
import ccxt
import pandas as pd
import statsmodels.api as sm
from datetime import datetime, timedelta
import os
import pickle
# 配置参数
TOKEN = "LAT" # 代币符号
TRAINING_HOURS = 24 # 训练数据的小时数
# 初始化 OKX 交易所
okx = ccxt.okx()
# 模型文件路径
MODEL_FILE_PATHS = {
"ols_price": "ols_price_model.pkl", # 价格预测模型
}
# 数据文件路径
DATA_FILE_PATHS = {
"second_level": "price_data_second.csv",
}
def download_data(token, trading_pair, training_hours, timeframe='1m'):
"""从 OKX 下载历史数据"""
end_time = datetime.now()
start_time = end_time - timedelta(hours=training_hours)
# 获取 OHLCV 数据
symbol = f"{token}/{trading_pair}" # 使用指定的交易对
ohlcv = okx.fetch_ohlcv(symbol, timeframe=timeframe, since=okx.parse8601(start_time.isoformat()))
# 转换为 DataFrame
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['date'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('date', inplace=True)
return df
def format_second_level_data(df):
"""格式化秒级数据"""
df = df.resample('1s').ffill() # 插值为秒级数据
df.to_csv(DATA_FILE_PATHS["second_level"])
return df
def train_ols_price_model():
"""训练价格预测 OLS 模型"""
df = pd.read_csv(DATA_FILE_PATHS["second_level"])
df['date'] = pd.to_datetime(df['date'])
df.set_index('date', inplace=True)
# 准备特征和目标变量
df['target'] = df['close'].shift(-1) # 目标变量是下一个时间点的收盘价
df.dropna(inplace=True)
# 特征变量
X = df[['open', 'high', 'low', 'close']]
X = sm.add_constant(X) # 添加常数项
# 目标变量
y = df['target']
# 训练 OLS 模型
model = sm.OLS(y, X)
results = model.fit()
# 保存模型
with open(MODEL_FILE_PATHS["ols_price"], "wb") as f:
pickle.dump(results, f)
print("OLS Price Model trained and saved.")
def update_data():
"""更新数据并训练模型"""
# 下载秒级数据（使用 LAT-USDT 交易对）
df_second = download_data(TOKEN, "USDT", TRAINING_HOURS, timeframe='1m')
format_second_level_data(df_second)
# 训练模型
train_ols_price_model()
def create_app():
"""创建 Flask 应用"""
app = Flask(__name__)
@app.route("/inference_price/<string:token>")
def generate_price_inference(token):
"""生成价格预测结果"""
if not token or token.upper() != TOKEN:
error_msg = "Token is required" if not token else "Token not supported"
return Response(json.dumps({"error": error_msg}), status=400, mimetype='application/json')
try:
# 加载模型
with open(MODEL_FILE_PATHS["ols_price"], "rb") as f:
model = pickle.load(f)
# 获取最新数据
df = download_data(token, "USDT", 1, timeframe='1m')
df = format_second_level_data(df)
# 准备特征变量
X_new = df[['open', 'high', 'low', 'close']].iloc[-1:]
X_new = sm.add_constant(X_new, has_constant='add')
# 预测下一个时间点的价格
forecast = model.predict(X_new)
return Response(str(forecast.iloc[0]), status=200)
except Exception as e:
return Response(json.dumps({"error": str(e)}), status=500, mimetype='application/json')
@app.route("/update")
def update():
"""更新数据并返回状态"""
try:
update_data()
return "0"
except Exception as e:
print(f"Error during update: {e}")
return "1"
return app
if __name__ == "__main__":
# 首次启动时更新数据并训练模型
if not all(os.path.exists(path) for path in MODEL_FILE_PATHS.values()):
update_data()
# 启动 Flask 应用，监听8001端口
app = create_app()
app.run(host="0.0.0.0", port=8001)
然后找一台特朗普送的 MacMini 就可以直接运行了。