How to Build an Autonomous Trading Agent with Python

📈 Crypto‑AI Mini‑Project

Goal: Build a tiny end‑to‑end application that pulls historic cryptocurrency price data, trains a lightweight AI model to forecast the next‑day price, and serves the prediction through a REST API that you can deploy to the cloud.

You’ll end up with a Docker‑containerised FastAPI service that:

1. Downloads the last 90 days of Bitcoin (BTC) price data from the free CoinGecko API.
2. Pre‑processes the data into a supervised learning problem (windowed time‑series).
3. Trains a Keras LSTM model (≈ 10 KB) on the data.
4. Exposes a single endpoint GET /predict?days=1 that returns the model’s forecast for the next n days.
5. Can be pushed to any container‑friendly host (Render, Railway, Fly.io, AWS ECS, etc.).

---

Table of Contents

Step	What you’ll do	Files created
0️⃣	Install prerequisites & set up a virtual environment	–
1️⃣	Create a new project folder & initialise Git	–
2️⃣	Pull historic price data (CoinGecko)	data_fetcher.py
3️⃣	Pre‑process & window the data	preprocess.py
4️⃣	Build & train the LSTM model	train.py
5️⃣	Save the trained model & test locally	model/
6️⃣	Wrap the model in a FastAPI service	app.py
7️⃣	Write Dockerfile & .dockerignore	Dockerfile, .dockerignore
8️⃣	Build & run the container locally	–
9️⃣	Deploy to a cloud provider (Render example)	–
🔟	(Optional) Add CI/CD with GitHub Actions	.github/workflows/docker.yml

---

0️⃣ Prerequisites

Tool	Why you need it	Install command / link
Python 3.10+	Core language	https://www.python.org/downloads/
Git	Version control	brew install git (mac) / sudo apt-get install git (Linux)
Docker	Containerisation & deployment	https://docs.docker.com/get-docker/
Virtualenv (or conda)	Isolate dependencies	python -m pip install --user virtualenv
API key (optional)	CoinGecko is free & no key needed, but you can swap with Binance/Coingecko‑Pro if you want higher rate limits.	–
Cloud account (Render, Railway, Fly.io, AWS, GCP…)	Host the container	Sign‑up for a free tier

Tip: The whole tutorial works on Windows, macOS, or Linux. The only OS‑specific part is Docker Desktop installation.

---

1️⃣ Project Setup

# 1️⃣ create a folder
mkdir crypto-ai-forecast && cd crypto-ai-forecast

# initialise git (optional but recommended)
git init

# 2️⃣ create a virtual environment
python -m venv .venv
# activate it
#   macOS / Linux:
source .venv/bin/activate
#   Windows:
.\.venv\Scripts\activate

# 3️⃣ upgrade pip & install core libs
python -m pip install --upgrade pip
pip install numpy pandas requests matplotlib scikit-learn tensorflow==2.13.0 fastapi uvicorn python-dotenv

Why those libs?

numpy/pandas – data wrangling.

requests – HTTP calls to CoinGecko.

matplotlib – quick sanity plot (optional).

scikit-learn – train‑test split & scaling.

tensorflow – LSTM model.

fastapi – lightweight ASGI API.

uvicorn – ASGI server.

python-dotenv – store secret env vars (e.g., future API keys).

Create a .gitignore now so you don’t commit the virtualenv or data files:

# .gitignore
.venv/
__pycache__/
*.pyc
*.pkl
*.h5
data/
model/
.env

---

2️⃣ Pull Historic Price Data

Create data_fetcher.py:

# data_fetcher.py
import requests
import pandas as pd
import os
from datetime import datetime, timedelta

COINGECKO_API = "https://api.coingecko.com/api/v3/coins/{id}/market_chart/range"

def fetch_price_history(coin_id: str = "bitcoin",
                       vs_currency: str = "usd",
                       days: int = 90) -> pd.DataFrame:
    """
    Pulls OHLCV data for the last `days` days.
    Returns a DataFrame with columns: ['timestamp', 'price'].
    """
    end = int(datetime.now().timestamp())
    start = int((datetime.now() - timedelta(days=days)).timestamp())
    url = COINGECKO_API.format(id=coin_id)

    params = {
        "vs_currency": vs_currency,
        "from": start,
        "to": end,
    }

    resp = requests.get(url, params=params, timeout=10)
    resp.raise_for_status()
    data = resp.json()

    # CoinGecko returns a list of [timestamp, price] pairs
    prices = pd.DataFrame(data["prices"], columns=["timestamp", "price"])
    # Convert ms → datetime
    prices["timestamp"] = pd.to_datetime(prices["timestamp"], unit="ms")
    prices.set_index("timestamp", inplace=True)
    # Resample to daily close price (last price of each day)
    daily = prices["price"].resample("D").last().ffill().reset_index()
    return daily

def main():
    df = fetch_price_history()
    os.makedirs("data", exist_ok=True)
    df.to_csv("data/btc_daily.csv", index=False)
    print(f"Saved {len(df)} rows to data/btc_daily.csv")

if __name__ == "__main__":
    main()

Run it:

python data_fetcher.py

You should now have data/btc_daily.csv looking like:

timestamp	price
2024‑05‑31	28 123
2024‑06‑01	28 456
…	…

---

3️⃣ Pre‑process & Window the Data

Create preprocess.py:

# preprocess.py
import pandas as pd
import numpy as np
from sklearn.preprocessing import MinMaxScaler
from pathlib import Path
import joblib   # pip install joblib

WINDOW_SIZE = 30   # past 30 days → predict next day

def load_data(csv_path: str = "data/btc_daily.csv") -> pd.DataFrame:
    return pd.read_csv(csv_path, parse_dates=["timestamp"])

def scale_series(series: pd.Series):
    scaler = MinMaxScaler()
    scaled = scaler.fit_transform(series.values.reshape(-1, 1))
    return scaled.squeeze(), scaler

def create_sequences(values: np.ndarray, window: int = WINDOW_SIZE):
    """
    Returns X shape (samples, window, 1) and y shape (samples, )
    """
    X, y = [], []
    for i in range(len(values) - window):
        X.append(values[i:i+window])
        y.append(values[i+window])
    X = np.array(X)
    y = np.array(y)
    # LSTM expects a feature dimension
    X = X[..., np.newaxis]   # (samples, window, 1)
    return X, y

def main():
    df = load_data()
    # Use only the price column
    price_series = df["price"]
    scaled_prices, scaler = scale_series(price_series)

    # Persist the scaler for later inverse‑transform
    Path("model").mkdir(parents=True, exist_ok=True)
    joblib.dump(scaler, "model/scaler.pkl")

    X, y = create_sequences(scaled_prices)
    # Train‑test split (80/20)
    split_idx = int(0.8 * len(X))
    X_train, X_test = X[:split_idx], X[split_idx:]
    y_train, y_test = y[:split_idx], y[split_idx:]

    # Save numpy arrays for quick loading in the training script
    np.save("model/X_train.npy", X_train)
    np.save("model/X_test.npy", X_test)
    np.save("model/y_train.npy", y_train)
    np.save("model/y_test.npy", y_test)

    print(f"Saved training data: {X_train.shape[0]} samples")
    print(f"Saved test data:    {X_test.shape[0]} samples")

if __name__ == "__main__":
    main()

Run it:

python preprocess.py

You now have:

model/
 ├─ X_train.npy
 ├─ X_test.npy
 ├─ y_train.npy
 ├─ y_test.npy
 └─ scaler.pkl

---

4️⃣ Build & Train the LSTM Model

Create train.py:

# train.py
import numpy as np
import tensorflow as tf
from tensorflow.keras import layers, callbacks, models
import joblib
from pathlib import Path

EPOCHS = 30
BATCH_SIZE = 16
MODEL_PATH = Path("model/btc_lstm.h5")

def load_data():
    X_train = np.load("model/X_train.npy")
    X_test = np.load("model/X_test.npy")
    y_train = np.load("model/y_train.npy")
    y_test = np.load("model/y_test.npy")
    return X_train, X_test, y_train, y_test

def build_model(input_shape):
    model = models.Sequential([
        layers.LSTM(64, activation='tanh', input_shape=input_shape, return_sequences=False),
        layers.Dense(32, activation='relu'),
        layers.Dense(1, activation='linear')
    ])
    model.compile(optimizer='adam', loss='mse')
    return model

def main():
    X_train, X_test, y_train, y_test = load_data()
    model = build_model(input_shape=(X_train.shape[1], 1))

    es = callbacks.EarlyStopping(patience=5, restore_best_weights=True)
    history = model.fit(
        X_train, y_train,
        validation_data=(X_test, y_test),
        epochs=EPOCHS,
        batch_size=BATCH_SIZE,
        callbacks=[es],
        verbose=2
    )
    # Save model
    MODEL_PATH.parent.mkdir(parents=True, exist_ok=True)
    model.save(MODEL_PATH)
    print(f"✅ Model saved to {MODEL_PATH}")

    # Quick evaluation (inverse‑scale to USD)
    scaler = joblib.load("model/scaler.pkl")
    preds = model.predict(X_test).flatten()
    mae = np.mean(np.abs(scaler.inverse_transform(preds[:, None]) -
                     scaler.inverse_transform(y_test[:, None])))
    print(f"Mean Absolute Error on test set: ${mae:,.2f}")

if __name__ == "__main__":
    main()

Run:

python train.py

You should see training loss decreasing, early stopping after a few epochs, and a final printed MAE (e.g., $150‑$300, which is fine for a demo).

The trained model (btc_lstm.h5) lives in model/.

---

5️⃣ Test the Model Locally

Create a tiny script predict_local.py to make sure everything works end‑to‑end:

# predict_local.py
import numpy as np
import joblib
import tensorflow as tf
import pandas as pd
from pathlib import Path

WINDOW_SIZE = 30
MODEL_PATH = Path("model/btc_lstm.h5")
SCALER_PATH = Path("model/scaler.pkl")
DATA_PATH = Path("data/btc_daily.csv")

def load_recent_window():
    df = pd.read_csv(DATA_PATH, parse_dates=["timestamp"])
    last_prices = df["price"].values[-WINDOW_SIZE:]   # last 30 days
    scaler = joblib.load(SCALER_PATH)
    scaled = scaler.transform(last_prices.reshape(-1, 1)).flatten()
    return scaled[np.newaxis, :, np.newaxis]   # shape (1,30,1)

def main():
    model = tf.keras.models.load_model(MODEL_PATH)
    scaler = joblib.load(SCALER_PATH)

    X = load_recent_window()
    pred_scaled = model.predict(X).flatten()
    pred_price = scaler.inverse_transform(pred_scaled[:, None]).flatten()[0]

    print(f"🔮 Next‑day BTC price forecast: ${pred_price:,.2f}")

if __name__ == "__main__":
    main()

Run it:

python predict_local.py

You should see something like:

🔮 Next‑day BTC price forecast: $30,172.45

Great! The model works.

---

6️⃣ Wrap the Model in a FastAPI Service

Create app.py:

# app.py
import joblib
import numpy as np
import tensorflow as tf
from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel
from pathlib import Path
import pandas as pd

app = FastAPI(title="Crypto‑AI Forecast API", version="0.1.0")

# -------------------------------------------------
# Load artefacts once at startup
# -------------------------------------------------
MODEL_PATH = Path("model/btc_lstm.h5")
SCALER_PATH = Path("model/scaler.pkl")
DATA_PATH = Path("data/btc_daily.csv")
WINDOW_SIZE = 30

model = tf.keras.models.load_model(MODEL_PATH)
scaler = joblib.load(SCALER_PATH)

def get_recent_window() -> np.ndarray:
    """Return the latest WINDOW_SIZE daily closes, scaled, shape (1,window,1)."""
    df = pd.read_csv(DATA_PATH, parse_dates=["timestamp"])
    if len(df) < WINDOW_SIZE:
        raise HTTPException(status_code=500,
                            detail=f"Not enough historic data (need {WINDOW_SIZE} days).")
    recent = df["price"].values[-WINDOW_SIZE:]
    scaled = scaler.transform(recent.reshape(-1, 1)).flatten()
    return scaled[np.newaxis, :, np.newaxis]

def predict_n_days(n: int = 1) -> list[float]:
    """
    Autoregressive forecast: repeatedly feed the model's prediction back
    as the newest time‑step to obtain n‑day ahead forecasts.
    """
    if n < 1:
        raise ValueError("n must be >= 1")
    window = get_recent_window()   # (1,30,1)

    preds = []
    for _ in range(n):
        nxt_scaled = model.predict(window).flatten()[0]
        nxt_price = scaler.inverse_transform(np.array([[nxt_scaled]])).flatten()[0]
        preds.append(float(nxt_price))

        # slide the window: drop first, append new scaled value
        window = np.append(window[:, 1:, :], [[[nxt_scaled]]], axis=1)
    return preds

# -------------------------------------------------
# API Endpoints
# -------------------------------------------------
class ForecastResponse(BaseModel):
    days_ahead: int
    forecast: list[float]

@app.get("/", tags=["root"])
def read_root():
    return {"message": "🚀 Crypto‑AI Forecast API. Use /predict?days=n"}

@app.get("/predict", response_model=ForecastResponse, tags=["prediction"])
def get_forecast(days: int = Query(1, ge=1, le=30, description="How many days ahead (max 30)")):
    """
    Returns a list of predicted BTC closing prices for the next *days* days.
    """
    try:
        forecasts = predict_n_days(days)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
    return ForecastResponse(days_ahead=days, forecast=forecasts)

Run the API locally

uvicorn app:app --host 0.0.0.0 --port 8000

Open a browser → http://localhost:8000/docs – you’ll see the automatically generated Swagger UI. Try GET /predict?days=3 and you should receive JSON like:

{
  "days_ahead": 3,
  "forecast": [30172.45, 30210.12, 30255.78]
}

---

7️⃣ Dockerise the Service

Create Dockerfile in the project root:

dockerfile
# Use the slim Python image (smaller attack surface)
FROM python:3.11-slim

# ----- Install OS dependencies (git, curl) -----
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# ----- Create a non‑root user -----
ARG UID=1000
ARG GID=1000
RUN groupadd -g $GID appgroup && \
    useradd -m -u $UID -g $GID -s /bin/bash appuser

# ----- Set workdir -----
WORKDIR /app

# ----- Copy only requirement files first (caching) -----
COPY requirements.txt .
RUN pip

#coding #tutorial #web3 #AI