🚀 End‑to‑End Tutorial
Build a decentralized AI‑powered crypto‑price‑prediction DApp
(Python‑ML model + Flask API + Chainlink External Adapter + Solidity contract + React front‑end)
You will end up with a test‑net DApp where users pay a small amount of ETH to request a 1‑hour‑ahead price prediction for a chosen crypto (e.g., BTC/USDT). The contract calls a Chainlink oracle that forwards the request to your off‑chain Flask service, which runs a pre‑trained LSTM model and returns the forecast.
📋 Table of Contents
| # | Section |
|---|---|
| 1 | Prerequisites & Tooling |
| 2 | Project Structure |
| 3 | 1️⃣ Data Collection & Exploration |
| 4 | 2️⃣ Train an LSTM price‑forecast model |
| 5 | 3️⃣ Export the model (ONNX) & Build a Flask API |
| 6 | 4️⃣ Chainlink External Adapter (Node.js) |
| 7 | 5️⃣ Solidity Smart Contract |
| 8 | 6️⃣ Deploy the contract on Sepolia (or any EVM testnet) |
| 9 | 7️⃣ Front‑end (React) – request a prediction |
| 10 | 8️⃣ Docker & CI/CD (optional) |
| 11 | 9️⃣ Security & Best‑Practices |
| 12 | 10️⃣ Next Steps & Extensions |
1️⃣ Prerequisites & Tooling
| Category | What you need | Why |
|---|---|---|
| Programming | Python ≥ 3.10, Node.js ≥ 18, npm/yarn, Solidity ≥ 0.8.19 | Core languages |
| Blockchain | MetaMask (browser), Sepolia ETH (test‑net faucet), Hardhat (or Foundry) | Deploy & interact |
| AI/ML |
pandas, numpy, scikit‑learn, torch, torchvision, onnx, onnxruntime
|
Model training & inference |
| Web | React ≥ 18, ethers.js, Vite (or CRA) | UI |
| DevOps | Docker ≥ 24, docker‑compose, Git | Containerisation |
| Chainlink | Chainlink node (free on Chainlink L2 testnet) or use Chainlink External Adapter (EA) sandbox | Oracle bridge |
| Optional | AWS / Railway / Fly.io (host Flask/EAs) | Public endpoint for the EA |
Tip: All commands below assume a Unix‑like shell (macOS/Linux). On Windows use WSL2 or Git‑Bash.
2️⃣ Project Structure
crypto‑ai‑dapp/
├─ contracts/
│ └─ PricePredictor.sol
├─ scripts/
│ └─ deploy.js # Hardhat deployment script
├─ src/
│ ├─ backend/
│ │ ├─ model/
│ │ │ └─ lstm.pt # PyTorch checkpoint
│ │ ├─ api/
│ │ │ └─ app.py # Flask API
│ │ └─ adapter/
│ │ └─ index.js # Chainlink EA (Node.js)
│ └─ frontend/
│ ├─ src/
│ │ ├─ App.jsx
│ │ └─ components/
│ └─ vite.config.ts
├─ docker-compose.yml
├─ Dockerfile.backend
├─ Dockerfile.adapter
├─ Dockerfile.frontend
└─ README.md
3️⃣ Data Collection & Exploration
We'll pull historical OHLCV data from Binance's REST API, store it locally as CSV, and do a quick sanity check.
3.1 Create a virtual env & install Python deps
cd crypto-ai-dapp/src/backend
python3 -m venv .venv
source .venv/bin/activate
pip install pandas numpy python-binance torch scikit-learn onnx onnxruntime
3.2 Data fetch script (fetch_data.py)
# src/backend/fetch_data.py
import pandas as pd
from binance.client import Client
import os, json, time
# Load Binance API key/secret from .env (optional – public endpoints work without)
API_KEY = os.getenv("BINANCE_API_KEY", "")
API_SECRET = os.getenv("BINANCE_API_SECRET", "")
client = Client(API_KEY, API_SECRET)
def fetch(symbol: str = "BTCUSDT", interval: str = "1h", lookback_days: int = 365):
"""
Pulls `lookback_days` of 1‑hour candles.
"""
limit = 1000 # max per request
end_ts = int(time.time() * 1000) # now in ms
start_ts = end_ts - lookback_days * 24 * 60 * 60 * 1000
all_klines = []
while start_ts < end_ts:
klines = client.get_historical_klines(
symbol, interval, start_str=str(start_ts), end_str=str(end_ts), limit=limit
)
if not klines:
break
all_klines.extend(klines)
start_ts = klines[-1][0] + 1 # next millisecond
time.sleep(0.2) # respect rate limits
df = pd.DataFrame(
all_klines,
columns=[
"open_time", "open", "high", "low", "close", "volume",
"close_time", "quote_asset_volume", "trades",
"taker_buy_base_asset_volume", "taker_buy_quote_asset_volume", "ignore"
],
)
df["open_time"] = pd.to_datetime(df["open_time"], unit="ms")
df["close_time"] = pd.to_datetime(df["close_time"], unit="ms")
numeric = ["open","high","low","close","volume"]
df[numeric] = df[numeric].astype(float)
return df
if __name__ == "__main__":
df = fetch()
df.to_csv("../data/btc_usdt_1h.csv", index=False)
print(f"Saved {len(df)} rows")
Run it once:
python fetch_data.py
You now have src/backend/data/btc_usdt_1h.csv. Open it in a notebook or pandas to verify.
3.3 Quick EDA (optional)
import pandas as pd
import matplotlib.pyplot as plt
df = pd.read_csv("../data/btc_usdt_1h.csv", parse_dates=["open_time"])
df.set_index("open_time", inplace=True)
df["close"].plot(figsize=(12,4), title="BTC/USDT 1‑h Close Price")
plt.show()
4️⃣ Train an LSTM Price‑Forecast Model
We'll predict the next hour’s close price given the previous 24 hours.
4.1 Prepare the dataset
# src/backend/train.py
import pandas as pd
import numpy as np
import torch
from torch import nn
from sklearn.preprocessing import MinMaxScaler
from pathlib import Path
DATA_PATH = Path(__file__).parent.parent / "data" / "btc_usdt_1h.csv"
df = pd.read_csv(DATA_PATH, parse_dates=["open_time"])
close = df["close"].values.reshape(-1, 1)
# Scale to [0,1]
scaler = MinMaxScaler()
close_scaled = scaler.fit_transform(close)
SEQ_LEN = 24 # look‑back window (24h)
PRED_LEN = 1 # predict next hour
def create_sequences(arr, seq_len, pred_len):
xs, ys = [], []
for i in range(len(arr) - seq_len - pred_len + 1):
xs.append(arr[i : i + seq_len])
ys.append(arr[i + seq_len : i + seq_len + pred_len])
return np.array(xs), np.array(ys)
X, y = create_sequences(close_scaled, SEQ_LEN, PRED_LEN)
# Train‑val split (80/20)
split = int(0.8 * len(X))
X_train, X_val = X[:split], X[split:]
y_train, y_val = y[:split], y[split:]
# Convert to torch tensors
X_train = torch.from_numpy(X_train).float()
y_train = torch.from_numpy(y_train).float()
X_val = torch.from_numpy(X_val).float()
y_val = torch.from_numpy(y_val).float()
4.2 Define the LSTM model
class PriceLSTM(nn.Module):
def __init__(self, input_dim=1, hidden_dim=64, num_layers=2, dropout=0.2):
super().__init__()
self.lstm = nn.LSTM(
input_dim, hidden_dim, num_layers,
batch_first=True, dropout=dropout, bidirectional=False
)
self.fc = nn.Linear(hidden_dim, 1) # output a single value
def forward(self, x):
# x shape: (batch, seq_len, input_dim)
out, _ = self.lstm(x)
# last hidden state
out = out[:, -1, :] # (batch, hidden_dim)
out = self.fc(out) # (batch, 1)
return out
4.3 Train loop
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = PriceLSTM().to(device)
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
EPOCHS = 30
BATCH = 64
def train_one_epoch():
model.train()
perm = torch.randperm(len(X_train))
epoch_loss = 0
for i in range(0, len(X_train), BATCH):
idx = perm[i : i + BATCH]
xb, yb = X_train[idx].to(device), y_train[idx].to(device)
optimizer.zero_grad()
pred = model(xb)
loss = criterion(pred, yb)
loss.backward()
optimizer.step()
epoch_loss += loss.item() * xb.size(0)
return epoch_loss / len(X_train)
def eval_one_epoch():
model.eval()
with torch.no_grad():
pred = model(X_val.to(device))
loss = criterion(pred, y_val.to(device))
return loss.item()
for epoch in range(1, EPOCHS + 1):
tr_loss = train_one_epoch()
val_loss = eval_one_epoch()
print(f"Epoch {epoch:02d} | train MSE: {tr_loss:.6f} | val MSE: {val_loss:.6f}")
# Save checkpoint & scaler
torch.save({
"model_state_dict": model.state_dict(),
"scaler": scaler
}, Path(__file__).parent / "model" / "lstm.pt")
print("✅ Model saved")
Run:
python train.py
You should see validation loss decreasing. If it plateaus, tweak hidden_dim, num_layers, or learning rate.
5️⃣ Export the Model (ONNX) & Build a Flask API
5.1 Convert to ONNX (framework‑agnostic)
# src/backend/export_onnx.py
import torch
from train import PriceLSTM, model, scaler, SEQ_LEN
import onnx
# Load checkpoint (if you rerun this script later)
checkpoint = torch.load("src/backend/model/lstm.pt", map_location="cpu")
model = PriceLSTM()
model.load_state_dict(checkpoint["model_state_dict"])
model.eval()
dummy_input = torch.randn(1, SEQ_LEN, 1) # batch=1
torch.onnx.export(
model,
dummy_input,
"src/backend/model/lstm.onnx",
input_names=["input"],
output_names=["output"],
dynamic_axes={"input": {0: "batch"}, "output": {0: "batch"}},
opset_version=15,
)
print("✅ ONNX model exported")
Run it once: python export_onnx.py.
Now lstm.onnx can be loaded in any runtime (e.g., onnxruntime in Flask).
5.2 Flask API (app.py)
# src/backend/api/app.py
import os
import json
import numpy as np
import pandas as pd
import onnxruntime as ort
from flask import Flask, request, jsonify
from sklearn.preprocessing import MinMaxScaler
app = Flask(__name__)
# Load model once at startup
MODEL_PATH = os.path.join(os.path.dirname(__file__), "..", "model", "lstm.onnx")
session = ort.InferenceSession(MODEL_PATH)
# Load scaler (saved with torch checkpoint)
import torch
ckpt = torch.load(os.path.join(os.path.dirname(__file__), "..", "model", "lstm.pt"), map_location="cpu")
scaler = ckpt["scaler"] # MinMaxScaler instance
SEQ_LEN = 24
def preprocess(series: np.ndarray) -> np.ndarray:
"""Scale and reshape for ONNX (batch, seq, 1)"""
scaled = scaler.transform(series.reshape(-1, 1))
return scaled.astype(np.float32).reshape(1, SEQ_LEN, 1)
@app.route("/predict", methods=["POST"])
def predict():
"""
Expected JSON:
{
"prices": [float, float, ..., float] # length = 24 (most recent close prices)
}
"""
data = request.get_json(force=True)
prices = data.get("prices")
if not prices or len(prices) != SEQ_LEN:
return jsonify({"error": f"`prices` must be a list of {SEQ_LEN} floats"}), 400
inp = preprocess(np.array(prices))
ort_inputs = {"input": inp}
ort_outs = session.run(None, ort_inputs)
pred_scaled = ort_outs[0] # shape (1,1)
# Inverse transform to original price scale
pred_price = scaler.inverse_transform(pred_scaled).flatten()[0]
return jsonify({"prediction": round(float(pred_price), 2)})
if __name__ == "__main__":
# For local dev only; in production use gunicorn / uvicorn
app.run(host="0.0.0.0", port=5000, debug=False)
Quick test
curl -X POST http://localhost:5000/predict \
-H "Content-Type: application/json" \
-d '{"prices": [30000,30120,30250, ... (24 values) ]}'
You should get {"prediction": 30412.34}.
5.3 Containerise the Flask API
Dockerfile.backend
# src/backend/Dockerfile.backend
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY model/ model/
COPY api/ api/
COPY train.py .
COPY export_onnx.py .
EXPOSE 5000
CMD ["python", "api/app.py"]
requirements.txt
flask
onnxruntime
numpy
pandas
scikit-learn
torch==2.3.0
Build & run locally:
docker build -t crypto-ai-backend -f src/backend/Dockerfile.backend .
docker run -p 5000:5000 crypto-ai-backend
6️⃣ Chainlink External Adapter (Node.js)
Chainlink nodes can call any HTTP endpoint. Instead of exposing the Flask service directly, we’ll wrap it in an External Adapter (EA) that follows Chainlink’s JSON‑RPC spec. This makes the integration easier when you spin up a free Chainlink node on Sepolia.
6.1 Initialise the EA project
cd src/backend/adapter
npm init -y
npm i express body-parser cors
Create index.js:
js
// src/backend/adapter/index.js
const express = require("express");
const bodyParser = require("body-parser");
const cors = require("cors");
const fetch = require("node-fetch"); // built‑in in Node 18+
const app = express();
app.use(cors());
app.use(bodyParser.json());
const FLASK_URL = process.env.FLASK_URL || "http://host.docker.internal:5000/predict";
/**
* Chainlink EA endpoint
* POST { id, data: { prices: [...] } }
*/
app.post("/", async (req, res) => {
const { id, data } = req.body;
if (!data?.prices || data.prices.length !== 24) {
return res.status(400).json({ jobRunID: id, statusCode: 400, error: "prices array of length 24 required" });
}
try {
const response = await fetch(FLASK_URL, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ prices: data.prices })
});
const json = await response.json();
// Chainlink expects { jobRunID, data: { result: ... } }
return res.json({
jobRunID: id,
statusCode: 200
#coding #tutorial #web3 #AI
Top comments (0)