How to Build an Autonomous Trading Agent with Python

🛠️ End‑to‑End Tutorial

Build an AI‑powered crypto‑price‑prediction service that logs every prediction on‑chain

You’ll end up with:

Component	Tech Stack	What it does
Data ingestion	Python + CoinGecko API	Pulls historic OHLCV data
Model	TensorFlow (LSTM)	Trains a short‑term price‑forecast model
API	FastAPI	Serves GET /predict and POST /log endpoints
Smart contract	Solidity (Ethereum)	Stores each prediction (timestamp, price) in an immutable ledger
Front‑end	React + Vite	Shows the latest prediction & lets users submit their own
Deployment	Docker Compose (local) → AWS ECS / GCP Cloud Run (optional)	One‑click spin‑up of the whole stack

---

1️⃣ Prerequisites

Category	Required
Programming	Python ≥ 3.10, JavaScript/TypeScript, basic Solidity
Tools	Git, Docker ≥ 20.10, Node ≥ 18, npm ≥ 9, VS Code (or any IDE)
Accounts	Free Infura or Alchemy project (Ethereum RPC), Etherscan API key (optional)
Crypto	Testnet ETH (e.g., Sepolia) – get via a faucet
Knowledge	1‑line basics of REST, LSTM, smart contracts, and Docker

Tip: If you’re new to any of these, skim the official “Getting Started” docs first – the tutorial works even if you only know the basics.

---

2️⃣ Project Layout

crypto‑ai‑predictor/
├─ backend/                # FastAPI + ML model
│   ├─ app/
│   │   ├─ main.py
│   │   ├─ model.py
│   │   └─ utils.py
│   ├─ Dockerfile
│   └─ requirements.txt
├─ contracts/              # Solidity contract + deployment scripts
│   ├─ PredictionLogger.sol
│   └─ scripts/
│       └─ deploy.ts
├─ frontend/               # React UI
│   ├─ src/
│   │   ├─ App.tsx
│   │   └─ api.ts
│   └─ Dockerfile
├─ docker-compose.yml
└─ README.md

We’ll fill each folder step‑by‑step.

---

3️⃣ Step‑by‑Step Implementation

3.1 Set up the repository

git clone https://github.com/yourname/crypto-ai-predictor.git
cd crypto-ai-predictor

Create a Python virtual environment (optional – Docker will handle it later):

python -m venv .venv
source .venv/bin/activate   # Windows: .venv\Scripts\activate

---

3.2 Backend – Data, Model, API

3.2.1 Install Python dependencies

Create backend/requirements.txt:

fastapi==0.110.0
uvicorn[standard]==0.29.0
pandas==2.2.2
numpy==1.26.4
scikit-learn==1.5.0
tensorflow==2.16.1
python-dotenv==1.0.1
requests==2.32.3
web3==6.19.0

pip install -r backend/requirements.txt

3.2.2 Pull historic price data

Create backend/app/utils.py:

import os
import requests
import pandas as pd
from datetime import datetime, timedelta

COINGECKO_API = "https://api.coingecko.com/api/v3"
DEFAULT_COIN = "bitcoin"
DEFAULT_VS_CURRENCY = "usd"

def fetch_ohlcv(days: int = 90, coin: str = DEFAULT_COIN) -> pd.DataFrame:
    """
    Returns a DataFrame with columns:
    ['timestamp','open','high','low','close','volume']
    """
    # CoinGecko returns daily candles for the last N days
    url = f"{COINGECKO_API}/coins/{coin}/ohlc"
    params = {"vs_currency": DEFAULT_VS_CURRENCY, "days": days}
    resp = requests.get(url, params=params)
    resp.raise_for_status()
    data = resp.json()  # [[unix, open, high, low, close], ...]

    df = pd.DataFrame(data, columns=["timestamp", "open", "high", "low", "close"])
    df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
    # Estimate volume (CoinGecko does not give it in the OHLC endpoint)
    # We'll fetch market data and calculate a proxy
    vol_url = f"{COINGECKO_API}/coins/{coin}/market_chart"
    vol_params = {"vs_currency": DEFAULT_VS_CURRENCY, "days": days}
    vol_resp = requests.get(vol_url, params=vol_params)
    vol_resp.raise_for_status()
    vol_data = vol_resp.json()["total_volumes"]  # [[unix, volume], ...]
    vol_df = pd.DataFrame(vol_data, columns=["timestamp", "volume"])
    vol_df["timestamp"] = pd.to_datetime(vol_df["timestamp"], unit="ms")
    df = df.merge(vol_df, on="timestamp")
    df.set_index("timestamp", inplace=True)
    return df

Explanation – CoinGecko’s free tier gives us 90‑day daily candles without any API key. The function merges volume data to give a full OHLCV dataset.

3.2.3 Build a simple LSTM model

Create backend/app/model.py:

import numpy as np
import pandas as pd
import tensorflow as tf
from tensorflow.keras import layers, models, callbacks
from sklearn.preprocessing import MinMaxScaler
from .utils import fetch_ohlcv

# ----------------------------------------------------------------------
# 1️⃣  Data preprocessing
# ----------------------------------------------------------------------
def prepare_dataset(df: pd.DataFrame, lookback: int = 30) -> tuple:
    """
    Returns (X, y) where:
      X shape = (samples, lookback, features)
      y shape = (samples, 1)   -> next day's closing price
    """
    scaler = MinMaxScaler()
    scaled = scaler.fit_transform(df)

    X, y = [], []
    for i in range(len(scaled) - lookback):
        X.append(scaled[i : i + lookback])
        y.append(scaled[i + lookback, 3])          # column 3 = close
    X = np.array(X)
    y = np.array(y).reshape(-1, 1)
    return X, y, scaler

# ----------------------------------------------------------------------
# 2️⃣  Model definition
# ----------------------------------------------------------------------
def build_lstm(input_shape):
    model = models.Sequential([
        layers.LSTM(64, activation='tanh', input_shape=input_shape),
        layers.Dense(32, activation='relu'),
        layers.Dense(1)  # predict scaled close price
    ])
    model.compile(optimizer='adam', loss='mse')
    return model

# ----------------------------------------------------------------------
# 3️⃣  Training routine (called from main)
# ----------------------------------------------------------------------
def train_and_save(model_path: str = "model.h5", lookback: int = 30):
    df = fetch_ohlcv(days=180)               # 6‑months of data for better generalisation
    X, y, scaler = prepare_dataset(df, lookback)

    model = build_lstm(input_shape=X.shape[1:])
    es = callbacks.EarlyStopping(patience=10, restore_best_weights=True)
    model.fit(X, y, epochs=200, batch_size=16, validation_split=0.2, callbacks=[es])

    # Save both model and scaler (pickle)
    model.save(model_path)
    import joblib, pathlib
    pathlib.Path("scaler.pkl").write_bytes(joblib.dumps(scaler))
    print(f"✅ Model saved to {model_path}")

Why LSTM? It captures temporal dependencies in price series with few parameters—perfect for a demo.

3.2.4 FastAPI server

Create backend/app/main.py:

import os
import json
import numpy as np
import pandas as pd
import joblib
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from tensorflow.keras.models import load_model
from .utils import fetch_ohlcv
from .model import prepare_dataset

app = FastAPI(title="Crypto AI Predictor", version="0.1.0")

# --------------------------------------------------------------
# Load model & scaler at startup
# --------------------------------------------------------------
MODEL_PATH = os.getenv("MODEL_PATH", "model.h5")
SCALER_PATH = os.getenv("SCALER_PATH", "scaler.pkl")
model = load_model(MODEL_PATH)
scaler = joblib.load(SCALER_PATH)

LOOKBACK = 30   # same as in training


class PredictResponse(BaseModel):
    timestamp: str
    predicted_price: float
    confidence: float | None = None   # placeholder for future extension


# --------------------------------------------------------------
# Helper: turn latest OHLCV into a prediction tensor
# --------------------------------------------------------------
def get_latest_tensor() -> np.ndarray:
    df = fetch_ohlcv(days=LOOKBACK + 1)   # we need LOOKBACK rows
    # Keep same column order as during training
    df = df[["open", "high", "low", "close", "volume"]]
    scaled = scaler.transform(df)
    tensor = scaled[-LOOKBACK:]            # shape (lookback, 5)
    return tensor.reshape((1, LOOKBACK, 5))


# --------------------------------------------------------------
# Public endpoint – return next‑day price prediction
# --------------------------------------------------------------
@app.get("/predict", response_model=PredictResponse)
def predict():
    try:
        tensor = get_latest_tensor()
        pred_scaled = model.predict(tensor)[0][0]          # scalar
        # Inverse‑scale only the *close* column (index 3)
        dummy = np.zeros((1, scaler.n_features_in_))
        dummy[0, 3] = pred_scaled
        pred_price = scaler.inverse_transform(dummy)[0, 3]
        ts = pd.Timestamp.utcnow().isoformat()
        return PredictResponse(timestamp=ts, predicted_price=round(float(pred_price), 2))
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


# --------------------------------------------------------------
# Optional: endpoint to trigger a re‑train (protected in prod)
# --------------------------------------------------------------
@app.post("/train")
def train():
    from .model import train_and_save
    train_and_save(model_path=MODEL_PATH)
    # reload
    global model, scaler
    model = load_model(MODEL_PATH)
    scaler = joblib.load(SCALER_PATH)
    return {"status": "retrained"}

Explanation –

• GET /predict grabs the most recent 30 days, scales them, feeds to the LSTM, then de‑scales the predicted close price.
• POST /train is a convenience for local testing; in production you’d protect it with an API key or CI pipeline.

3.2.5 Dockerize the backend

Create backend/Dockerfile:

# syntax = docker/dockerfile:1.4
FROM python:3.12-slim AS builder

WORKDIR /app
COPY backend/requirements.txt .
RUN pip install --upgrade pip && \
    pip install --no-cache-dir -r requirements.txt

# ---- Runtime image ----
FROM python:3.12-slim
WORKDIR /app
COPY --from=builder /usr/local/lib/python3.12/site-packages /usr/local/lib/python3.12/site-packages
COPY backend/app ./app

# Model artefacts (you can also mount them as volumes)
COPY backend/model.h5 .
COPY backend/scaler.pkl .

EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

Why multi‑stage? Keeps the final image tiny (~70 MB) – perfect for serverless containers.

---

3.3 Smart Contract – Log Predictions On‑Chain

3.3.1 Solidity contract

Create contracts/PredictionLogger.sol:

// SPDX-License-Identifier: MIT
pragma solidity ^0.8.24;

contract PredictionLogger {
    struct Prediction {
        uint256 timestamp;   // block timestamp when logged
        uint256 price;       // price * 1e2 (e.g., $27,345.12 → 2734512)
        address reporter;   // who logged it
    }

    Prediction[] public predictions;

    event PredictionLogged(uint256 indexed idx, uint256 timestamp, uint256 price, address reporter);

    /// @notice Store a new prediction. Caller pays only gas.
    /// @param price price in cents (2 decimals) to avoid floating points.
    function logPrediction(uint256 price) external {
        predictions.push(Prediction({
            timestamp: block.timestamp,
            price: price,
            reporter: msg.sender
        }));
        emit PredictionLogged(predictions.length - 1, block.timestamp, price, msg.sender);
    }

    /// @notice Get total number of predictions logged.
    function count() external view returns (uint256) {
        return predictions.length;
    }

    /// @notice Retrieve a prediction by index.
    function get(uint256 idx) external view returns (Prediction memory) {
        require(idx < predictions.length, "out of range");
        return predictions[idx];
    }
}

Design notes

• We store price as an integer with 2 decimal places (uint256 price). Solidity has no floating‑point numbers.
• logPrediction is payable‑free – anyone can call it, but you could add an onlyOwner or a small fee later.

3.3.2 Hardhat setup (deployment script)

cd contracts
npm init -y
npm i --save-dev hardhat @nomicfoundation/hardhat-toolbox ethers dotenv
npx hardhat
# Choose "Create a basic sample project"

Add .env in contracts/:

SEPOLIA_RPC_URL=https://sepolia.infura.io/v3/YOUR_INFURA_PROJECT_ID
PRIVATE_KEY=0xYOUR_PRIVATE_KEY   # account with testnet ETH

Update hardhat.config.ts:

import { config as dotenvConfig } from "dotenv";
import { HardhatUserConfig } from "hardhat/types";
dotenvConfig();

const config: HardhatUserConfig = {
  solidity: "0.8.24",
  networks: {
    sepolia: {
      url: process.env.SEPOLIA_RPC_URL || "",
      accounts: process.env.PRIVATE_KEY ? [process.env.PRIVATE_KEY] : [],
    },
  },
};

export default config;

Create scripts/deploy.ts:

import { ethers } from "hardhat";

async function main() {
  const PredictionLogger = await ethers.getContractFactory("PredictionLogger");
  const logger = await PredictionLogger.deploy();
  await logger.waitForDeployment();

  console.log("✅ PredictionLogger deployed to:", await logger.getAddress());
}

main().catch((error) => {
  console.error(error);
  process.exitCode = 1;
});

Deploy to Sepolia:

npx hardhat run scripts/deploy.ts --network sepolia
# copy the printed address – you’ll need it in the backend

3.3.3 ABI export

After compilation, the ABI is in artifacts/contracts/PredictionLogger.sol/PredictionLogger.json. Copy the abi array into the backend folder:

mkdir -p backend/app/abi
cp artifacts/contracts/PredictionLogger.sol/PredictionLogger.json backend/app/abi/

Rename it to PredictionLogger_abi.json for clarity.

---

3.4 Connect FastAPI to the Contract

Add web3 to the backend (already in requirements.txt).

Create backend/app/blockchain.py:

python
import os
from web3 import Web3
import json
from pathlib import Path

# Load env variables (you can use python-dotenv)
INFURA_URL = os.getenv("INFURA_URL")  # e.g. https://sepolia.infura.io/v3/xxxx
PRIVATE_KEY = os.getenv("PRIVATE_KEY")  # for signing txs (test account)

w3 = Web3(Web3.HTTPProvider(INFURA_URL))
assert w3.is_connected(), "❌ Can't connect to Ethereum node"

# Load contract ABI & address
ABI_PATH = Path(__file__).parent / "abi" / "PredictionLogger_abi.json"
with open(ABI_PATH) as f:
    abi = json.load(f)["abi"]

CONTRACT_ADDRESS = os.getenv("CONTRACT_ADDRESS")  # set after deployment
contract = w3.eth.contract(address=CONTRACT_ADDRESS, abi=abi)

def log_prediction_onchain(price_usd: float) -> str:
    """
    Sends a transaction to store `price_usd` (2 decimal precision) on‑chain.
    Returns the transaction hash.
    """
    # Convert to integer cents
    price_cents = int(round(price_usd * 100))

    # Build transaction
    nonce = w3.eth.get_transaction_count(w3.eth.account.from_key(PRIVATE_KEY).address)
    tx = contract.functions.logPrediction(price_cents).build_transaction({
        "chainId": w3.eth.chain_id,
        "gas": 200_000,
        "gasPrice": w3.to_wei("5", "gwei"),


#coding #tutorial #web3 #AI