I've spent the last six months building quantitative trading systems across DeFi protocols and centralized exchanges, and I can tell you firsthand that the gap between a strategy idea and a production-ready backtester is where most traders give up. After cycling through multiple frameworks—Zipline, VectorBT, and custom solutions—I keep returning to Backtrader combined with Tardis.dev as my go-to stack for historical data analysis and strategy prototyping. In this guide, I'll show you exactly how to wire these tools together, integrate HolySheep AI for intelligent signal generation, and benchmark costs so you can make data-driven decisions before spending a single dollar on API credits.
Why Backtrader + Tardis.dev + HolySheep AI
The cryptocurrency markets never sleep, and neither should your backtesting pipeline. Backtrader remains the most Pythonic and extensible open-source backtesting framework available, with native support for multiple data feeds, broker simulators, and analyzers. Tardis.dev provides institutional-grade historical market data—including trades, order books, funding rates, and liquidations—for Binance, Bybit, OKX, Deribit, and 30+ other exchanges with a unified API. HolySheep AI brings large language model inference directly into your strategy logic at a fraction of the cost: DeepSeek V3.2 at $0.42 per million output tokens versus $8.00 for GPT-4.1 or $15.00 for Claude Sonnet 4.5.
For a typical quantitative workload of 10 million tokens per month (common when generating daily signals across 20+ trading pairs with multi-turn reasoning), the cost difference is staggering: $4.20/month with DeepSeek V3.2 versus $80/month with GPT-4.1 or $150/month with Claude Sonnet 4.5. That's 95% savings—money you can reinvest into data subscriptions or hardware.
2026 AI Inference Cost Comparison
| Provider / Model | Output Price ($/MTok) | 10M Tokens/Month | 100M Tokens/Month | Latency |
|---|---|---|---|---|
| HolySheep AI - DeepSeek V3.2 | $0.42 | $4.20 | $42.00 | <50ms |
| HolySheep AI - Gemini 2.5 Flash | $2.50 | $25.00 | $250.00 | <80ms |
| HolySheep AI - GPT-4.1 | $8.00 | $80.00 | $800.00 | <120ms |
| HolySheep AI - Claude Sonnet 4.5 | $15.00 | $150.00 | $1,500.00 | <150ms |
Prerequisites and Environment Setup
Before diving into code, ensure you have Python 3.10+ installed. I recommend using a virtual environment to avoid dependency conflicts. The complete stack requires only a handful of packages:
python -m venv backtest_env
source backtest_env/bin/activate # Linux/macOS
backtest_env\Scripts\activate # Windows
pip install backtrader pandas numpy requests
pip install TardisClient # Official Python SDK for Tardis.dev
pip install python-dotenv # For secure API key managementPart 1: Fetching Historical Data from Tardis.dev via HolySheep AI
Tardis.dev offers both REST and WebSocket APIs for historical data retrieval. For backtesting, you'll primarily use the REST API to download aggregated candle (OHLCV) data or raw trade streams. HolySheep AI's relay infrastructure routes your requests through optimized global endpoints, reducing latency when pulling large datasets. The base URL for all HolySheep AI API calls is https://api.holysheep.ai/v1.
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
HolySheep AI Configuration
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
Tardis.dev Configuration
TARDIS_API_KEY = os.getenv("TARDIS_API_KEY", "YOUR_TARDIS_API_KEY")
TARDIS_EXCHANGE = "binance" # Options: binance, bybit, okx, deribit, etc.
TARDIS_SYMBOL = "BTC-USDT"
TARDIS_START_DATE = "2025-01-01"
TARDIS_END_DATE = "2025-12-31"
HolySheep AI Model Selection
Cost-efficient: deepseek-v3-250324 (DeepSeek V3.2)
Balanced: gemini-2.5-flash (Gemini 2.5 Flash)
Premium: gpt-4.1 or claude-sonnet-4.5-20250514
HOLYSHEEP_MODEL = "deepseek-v3-250324"The .env file should contain your credentials:
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
TARDIS_API_KEY=YOUR_TARDIS_API_KEYPart 2: Fetching OHLCV Data from Tardis.dev
Tardis.dev provides historical OHLCV data through their market-ohlcv endpoint. The following script downloads daily candles for a specified date range, handling pagination automatically for multi-month requests.
# fetch_tardis_data.py
import requests
import pandas as pd
from datetime import datetime, timedelta
from config import TARDIS_API_KEY, TARDIS_EXCHANGE, TARDIS_SYMBOL
def fetch_ohlcv_data(
exchange: str,
symbol: str,
start_date: str,
end_date: str,
timeframe: str = "1d",
limit: int = 1000
) -> pd.DataFrame:
"""
Fetch OHLCV data from Tardis.dev API.
Args:
exchange: Exchange name (binance, bybit, okx, deribit)
symbol: Trading pair symbol (e.g., BTC-USDT)
start_date: Start date in YYYY-MM-DD format
end_date: End date in YYYY-MM-DD format
timeframe: Candle timeframe (1m, 5m, 1h, 1d)
limit: Maximum records per request (max 1000 for Tardis)
Returns:
DataFrame with columns: timestamp, open, high, low, close, volume
"""
base_url = f"https://api.tardis.dev/v1/feeds/{exchange}:{symbol}"
params = {
"from": start_date,
"to": end_date,
"timeframe": timeframe,
"limit": limit,
}
headers = {"Authorization": f"Bearer {TARDIS_API_KEY}"}
all_candles = []
page_token = None
print(f"Fetching {timeframe} OHLCV data for {symbol} on {exchange}...")
print(f"Date range: {start_date} to {end_date}")
while True:
if page_token:
params["page_token"] = page_token
response = requests.get(base_url, params=params, headers=headers, timeout=30)
response.raise_for_status()
data = response.json()
# Parse OHLCV data (Tardis returns array of [timestamp, open, high, low, close, volume])
candles = data.get("data", [])
if not candles:
break
for candle in candles:
all_candles.append({
"timestamp": pd.to_datetime(candle[0], unit="ms"),
"open": float(candle[1]),
"high": float(candle[2]),
"low": float(candle[3]),
"close": float(candle[4]),
"volume": float(candle[5]),
})
print(f" Fetched {len(candles)} candles... Total: {len(all_candles)}")
# Handle pagination
page_token = data.get("next_page_token")
if not page_token:
break
df = pd.DataFrame(all_candles)
df.set_index("timestamp", inplace=True)
df.sort_index(inplace=True)
print(f"Total candles retrieved: {len(df)}")
print(f"Date range: {df.index.min()} to {df.index.max()}")
print(f"Data shape: {df.shape}")
return df
if __name__ == "__main__":
# Example: Fetch 1-year daily BTC/USDT data
df = fetch_ohlcv_data(
exchange=TARDIS_EXCHANGE,
symbol=TARDIS_SYMBOL,
start_date=TARDIS_START_DATE,
end_date=TARDIS_END_DATE,
timeframe="1d"
)
# Save to CSV for Backtrader consumption
df.to_csv("btc_usdt_daily.csv")
print("Data saved to btc_usdt_daily.csv")Part 3: Integrating HolySheep AI for Strategy Signal Generation
This is where the magic happens. Instead of hardcoding entry/exit rules, you can leverage HolySheep AI to analyze price action, volume patterns, and market microstructure—and generate trading signals dynamically. The following class wraps the HolySheep AI chat completions API with streaming support, token counting, and cost tracking.
# holy_sheep_client.py
import requests
import tiktoken
import time
from typing import Generator, Optional, List, Dict, Any
from config import HOLYSHEEP_API_KEY, HOLYSHEEP_BASE_URL, HOLYSHEEP_MODEL
class HolySheepAIClient:
"""
Python client for HolySheep AI API with cost tracking and streaming support.
HolySheep AI offers 85%+ savings vs official pricing:
- Rate: $1 USD = ¥7.3 CNY (saves 85%+ vs ¥7.3)
- Payment: WeChat, Alipay supported
- Latency: <50ms for most requests
- Free credits on signup
"""
def __init__(self, api_key: str, model: str = "deepseek-v3-250324"):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
self.model = model
self.total_input_tokens = 0
self.total_output_tokens = 0
self.total_cost = 0.0
# Pricing per million tokens (output)
self.pricing = {
"deepseek-v3-250324": 0.42, # DeepSeek V3.2: $0.42/MTok
"gemini-2.5-flash": 2.50, # Gemini 2.5 Flash: $2.50/MTok
"gpt-4.1": 8.00, # GPT-4.1: $8.00/MTok
"claude-sonnet-4.5-20250514": 15.00 # Claude Sonnet 4.5: $15.00/MTok
}
def estimate_tokens(self, text: str) -> int:
"""Estimate token count using cl100k_base encoding ( approximates GPT-4 tokenizer )."""
try:
encoding = tiktoken.get_encoding("cl100k_base")
return len(encoding.encode(text))
except Exception:
# Fallback: roughly 4 characters per token
return len(text) // 4
def chat(
self,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: int = 500,
stream: bool = False
) -> Dict[str, Any]:
"""
Send a chat completion request to HolySheep AI.
Args:
messages: List of message dicts with 'role' and 'content'
temperature: Sampling temperature (0.0-2.0)
max_tokens: Maximum output tokens
stream: Enable streaming response
Returns:
Dict with 'content', 'usage', and 'cost' fields
"""
endpoint = f"{self.base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self.model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
"stream": stream
}
start_time = time.time()
if stream:
return self._stream_request(endpoint, headers, payload, start_time)
else:
return self._sync_request(endpoint, headers, payload, start_time)
def _sync_request(self, endpoint: str, headers: dict, payload: dict, start_time: float) -> Dict[str, Any]:
"""Handle synchronous (non-streaming) requests."""
response = requests.post(endpoint, headers=headers, json=payload, timeout=30)
response.raise_for_status()
data = response.json()
latency_ms = (time.time() - start_time) * 1000
usage = data.get("usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
# Calculate cost
cost_per_token = self.pricing.get(self.model, 0.42) / 1_000_000
cost = output_tokens * cost_per_token
# Update totals
self.total_input_tokens += input_tokens
self.total_output_tokens += output_tokens
self.total_cost += cost
return {
"content": data["choices"][0]["message"]["content"],
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"total_tokens": input_tokens + output_tokens,
"cost": cost,
"latency_ms": round(latency_ms, 2),
"model": self.model
}
def _stream_request(self, endpoint: str, headers: dict, payload: dict, start_time: float) -> Generator[str, None, None]:
"""Handle streaming requests."""
response = requests.post(endpoint, headers=headers, json=payload, stream=True, timeout=60)
full_content = []
for line in response.iter_lines():
if line:
line_text = line.decode("utf-8")
if line_text.startswith("data: "):
if line_text.strip() == "data: [DONE]":
break
chunk_data = line_text[6:] # Remove "data: " prefix
import json
try:
chunk = json.loads(chunk_data)
delta = chunk["choices"][0]["delta"].get("content", "")
if delta:
full_content.append(delta)
yield delta
except json.JSONDecodeError:
continue
# Update totals after streaming completes
combined = "".join(full_content)
output_tokens = self.estimate_tokens(combined)
cost = output_tokens * self.pricing.get(self.model, 0.42) / 1_000_000
self.total_output_tokens += output_tokens
self.total_cost += cost
def get_cost_report(self) -> Dict[str, Any]:
"""Return a summary of token usage and costs."""
return {
"model": self.model,
"total_input_tokens": self.total_input_tokens,
"total_output_tokens": self.total_output_tokens,
"total_cost_usd": round(self.total_cost, 4),
"estimated_savings_vs_gpt4": round(
(8.00 - self.pricing.get(self.model, 0.42)) * self.total_output_tokens / 1_000_000, 2
)
}
Example usage
if __name__ == "__main__":
client = HolySheepAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
model="deepseek-v3-250324" # Most cost-efficient model
)
messages = [
{
"role": "system",
"content": "You are a cryptocurrency technical analyst. Analyze the provided OHLCV data and output a JSON signal: {'action': 'buy'|'sell'|'hold', 'confidence': 0.0-1.0, 'reason': 'brief explanation'}"
},
{
"role": "user",
"content": "BTC/USDT daily candles for the past 7 days: Close prices were 67250, 68100, 68900, 68500, 69200, 70100, 71500. Current volume is 20% above 30-day average. What is your trading signal?"
}
]
result = client.chat(messages, temperature=0.3, max_tokens=200)
print(f"Signal: {result['content']}")
print(f"Tokens used: {result['output_tokens']}")
print(f"Cost: ${result['cost']:.4f}")
print(f"Latency: {result['latency_ms']}ms")
print(f"\nTotal cost report: {client.get_cost_report()}")Part 4: Building a Complete Backtrader Strategy with AI Signals
Now we wire everything together. The AISignalStrategy class extends Backtrader's Strategy base class, feeds in historical OHLCV data, queries HolySheep AI at configurable intervals (e.g., every 7 candles or weekly), and executes trades based on the returned signals.
# ai_backtrader_strategy.py
import backtrader as bt
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from holy_sheep_client import HolySheepAIClient
from config import HOLYSHEEP_API_KEY, HOLYSHEEP_MODEL
class AISignalStrategy(bt.Strategy):
"""
Backtrader strategy that generates trading signals using HolySheep AI.
The strategy analyzes recent price action, volume, and volatility,
then queries the AI model to produce buy/sell/hold recommendations.
HolySheep AI provides:
- 85%+ cost savings vs official APIs
- DeepSeek V3.2 at $0.42/MTok output
- <50ms latency for real-time inference
- WeChat/Alipay payment options
"""
params = (
("ai_model", HOLYSHEEP_MODEL),
("signal_interval", 7), # Query AI every N candles
("confidence_threshold", 0.65), # Minimum confidence to execute
("position_size", 0.95), # Use 95% of available capital
("stop_loss", 0.05), # 5% stop-loss
("take_profit", 0.10), # 10% take-profit
("max_hold_days", 14), # Maximum holding period
("verbose", True),
)
def __init__(self):
self.ai_client = HolySheepAIClient(
api_key=HOLYSHEEP_API_KEY,
model=self.params.ai_model
)
self.candle_count = 0
self.last_signal = None
self.entry_price = None
self.entry_date = None
self.order = None
self.trade_log = []
# Indicators
self.sma20 = bt.indicators.SimpleMovingAverage(self.data.close, period=20)
self.sma50 = bt.indicators.SimpleMovingAverage(self.data.close, period=50)
self.rsi = bt.indicators.RSI(self.data.close, period=14)
self.bb = bt.indicators.BollingerBands(self.data.close, period=20, devfactor=2)
def _build_ai_prompt(self) -> str:
"""Construct the AI analysis prompt from current market data."""
lookback = 14 # Analyze last 14 candles
closes = [f"{i+1}d ago: ${c:.2f}" for i, c in enumerate(
reversed([self.data.close[-i] for i in range(1, min(lookback+1, len(self.data)-1))]))
]
volumes = [f"{i+1}d ago: {v:.0f}" for i, v in enumerate(
reversed([self.data.volume[-i] for i in range(1, min(lookback+1, len(self.data)-1))]))
]
prompt = f"""Analyze this cryptocurrency market data and provide a trading signal.
CURRENT MARKET DATA:
- Symbol: {self.data._name if hasattr(self.data, '_name') else 'UNKNOWN'}
- Current Price: ${self.data.close[0]:.2f}
- 20-day SMA: ${self.sma20[0]:.2f}
- 50-day SMA: ${self.sma50[0]:.2f}
- RSI(14): {self.rsi[0]:.1f}
- Bollinger Upper: ${self.bb.lines.top[0]:.2f}
- Bollinger Lower: ${self.bb.lines.bot[0]:.2f}
- Current Position: {'LONG' if self.position.size > 0 else 'FLAT'}
RECENT CLOSING PRICES (oldest to newest):
{chr(10).join(closes)}
RECENT VOLUMES (oldest to newest):
{chr(10).join(volumes)}
OUTPUT FORMAT (respond ONLY with valid JSON):
{{"action": "buy", "confidence": 0.75, "reason": "RSI oversold with strong volume"}}
Rules:
- action: "buy" (strong bullish signal), "sell" (strong bearish signal), or "hold" (no clear opportunity)
- confidence: 0.0 to 1.0 (higher = more confident)
- Only output valid JSON, no markdown or explanation"""
return prompt
def _query_ai_signal(self) -> dict:
"""Query HolySheep AI for a trading signal."""
messages = [
{"role": "system", "content": "You are an expert cryptocurrency quantitative analyst."},
{"role": "user", "content": self._build_ai_prompt()}
]
try:
result = self.ai_client.chat(
messages,
temperature=0.3,
max_tokens=150,
stream=False
)
import json
# Parse JSON response
content = result["content"].strip()
if content.startswith("```json"):
content = content[7:]
if content.startswith("```"):
content = content[3:]
if content.endswith("```"):
content = content[:-3]
signal = json.loads(content.strip())
signal["cost"] = result["cost"]
signal["latency_ms"] = result["latency_ms"]
signal["tokens"] = result["output_tokens"]
return signal
except json.JSONDecodeError as e:
if self.params.verbose:
print(f" [AI PARSE ERROR] Failed to parse AI response: {e}")
return {"action": "hold", "confidence": 0.0, "reason": "AI parse error"}
except Exception as e:
if self.params.verbose:
print(f" [AI API ERROR] {type(e).__name__}: {e}")
return {"action": "hold", "confidence": 0.0, "reason": f"API error: {str(e)[:50]}"}
def _execute_trade(self, signal: dict):
"""Execute a trade based on the AI signal."""
action = signal["action"]
confidence = signal["confidence"]
reason = signal.get("reason", "No reason provided")
if self.order:
return # Already have pending order
if action == "buy" and confidence >= self.params.confidence_threshold:
if self.position.size == 0:
size = self.broker.getcash() * self.params.position_size / self.data.close[0]
self.order = self.buy(size=size)
self.entry_price = self.data.close[0]
self.entry_date = self.datas[0].datetime.date(0)
if self.params.verbose:
print(f" [BUY] Size: {size:.4f} @ ${self.entry_price:.2f} | Confidence: {confidence:.2f}")
print(f" Reason: {reason}")
elif action == "sell" and confidence >= self.params.confidence_threshold:
if self.position.size > 0:
self.order = self.close()
if self.params.verbose:
pnl = (self.data.close[0] - self.entry_price) / self.entry_price * 100
print(f" [SELL] @ ${self.data.close[0]:.2f} | PnL: {pnl:+.2f}%")
print(f" Reason: {reason}")
def notify_order(self, order):
"""Handle order notifications."""
if order.status in [order.Completed]:
if order.isbuy():
self.trade_log.append({
"date": self.datas[0].datetime.date(0),
"action": "BUY",
"price": order.executed.price,
"size": order.executed.size,
"value": order.executed.value
})
elif order.issell():
self.trade_log.append({
"date": self.datas[0].datetime.date(0),
"action": "SELL",
"price": order.executed.price,
"size": order.executed.size,
"value": order.executed.value
})
self.order = None
def next(self):
"""Main strategy logic executed on each new candle."""
self.candle_count += 1
# Query AI at specified intervals
if self.candle_count % self.params.signal_interval == 0:
if self.params.verbose:
print(f"\n[{self.data.datetime.date(0)}] Candle #{self.candle_count}")
print(f" Price: ${self.data.close[0]:.2f} | SMA20: ${self.sma20[0]:.2f} | RSI: {self.rsi[0]:.1f}")
# Check max hold time
if self.position.size > 0 and self.entry_date:
hold_days = (self.datas[0].datetime.date(0) - self.entry_date).days
if hold_days >= self.params.max_hold_days:
if self.params.verbose:
print(f" [TIMEOUT] Max hold period ({self.params.max_hold_days}d) reached. Closing position.")
self.order = self.close()
return
# Query AI
signal = self._query_ai_signal()
self.last_signal = signal
if self.params.verbose:
print(f" [AI SIGNAL] Action: {signal['action']} | Confidence: {signal['confidence']:.2f}")
print(f" Cost: ${signal.get('cost', 0):.4f} | Latency: {signal.get('latency_ms', 0)}ms")
# Execute trade
self._execute_trade(signal)
# Stop-loss check (runs every candle)
if self.position.size > 0 and self.entry_price:
drawdown = (self.entry_price - self.data.close[0]) / self.entry_price
if drawdown >= self.params.stop_loss:
if self.params.verbose:
print(f" [STOP-LOSS] Drawdown: {drawdown*100:.2f}% | Closing position")
self.order = self.close()
# Take-profit check
if self.position.size > 0 and self.entry_price:
profit = (self.data.close[0] - self.entry_price) / self.entry_price
if profit >= self.params.take_profit:
if self.params.verbose:
print(f" [TAKE-PROFIT] Profit: {profit*100:.2f}% | Taking profits")
self.order = self.close()
def stop(self):
"""Called when backtesting ends. Print summary."""
cost_report = self.ai_client.get_cost_report()
print(f"\n{'='*60}")
print(f"BACKTEST COMPLETE")
print(f"{'='*60}")
print(f"Total AI calls: {self.candle_count // self.params.signal_interval}")
print(f"Total trades: {len(self.trade_log)}")
print(f"Total AI cost: ${cost_report['total_cost_usd']:.4f}")
print(f"Total output tokens: {cost_report['total_output_tokens']:,}")
print(f"Model: {cost_report['model']}")
print(f"Final portfolio value: ${self.broker.getvalue():.2f}")
print(f"{'='*60}")Part 5: Running the Complete Backtest
With the data fetcher, AI client, and strategy in place, the main execution script ties everything together, configures the Backtrader engine, and runs the simulation.
# run_backtest.py
import backtrader as bt
import pandas as pd
from datetime import datetime
from ai_backtrader_strategy import AISignalStrategy
from fetch_tardis_data import fetch_ohlcv_data
from config import (
TARDIS_EXCHANGE, TARDIS_SYMBOL,
TARDIS_START_DATE, TARDIS_END_DATE,
HOLYSHEEP_MODEL
)
class TardisCSVData(bt.feeds.GenericCSVData):
"""Custom Backtrader data feed for Tardis-exported CSV files."""
params = (
("dtformat", "%Y-%m-%d %H:%M:%S"),
("datetime", 0),
("open", 1),
("high", 2),
("low", 3),
("close", 4),
("volume", 5),
("openinterest", -1),
("header", True),
)
class PandasData(bt.feeds.PandasData):
"""Backtrader data feed from pandas DataFrame."""
params = (
("datetime", None),
("open", "open"),
("high", "high"),
("low", "low"),
("close", "close"),
("volume", "volume"),
("openinterest", -1),
)
def run_backtest(
symbol: str = "BTC-USDT",
exchange: str = "binance",
start_date: str = "2025-01-01",
end_date: str = "2025-12-31",
initial_cash: float = 100000.0,
ai_model: str = "deepseek-v3-250324"
):
"""
Run a complete backtest with Tardis data and HolySheep AI signals.
Args:
symbol: Trading pair symbol (e.g., BTC-USDT)
exchange: Exchange name (binance, bybit, okx, deribit)
start_date: Backtest start date (YYYY-MM-DD)
end_date: Backtest end date (YYYY-MM-DD)
initial_cash: Starting capital in USD
ai_model: HolySheep AI model to use
Returns:
Backtrader Cerebro instance with results
"""
print(f"\n{'='*70}")
print(f"HOLYSHEEP AI + BACKTRADER + TARDIS.DEV CRYPTO BACKTEST")
print(f"{'='*70}")
print(f"Symbol: {symbol}")
print(f"Exchange: {exchange}")
print(f"Period: {start_date} to {end_date}")
print(f"Initial Capital: ${initial_cash:,.2f}")
print(f"AI Model: {ai_model}")
print(f"{'='*70}\n")
# Step 1: Fetch historical data from Tardis.dev
print("Step 1: Fetching data from Tardis.dev...")
csv_filename = f"{symbol.replace('-', '_')}_{exchange}.csv"
try:
df = fetch_ohlcv_data(
exchange=exchange,
symbol=symbol,
start_date=start_date,
end_date=end_date,
timeframe="1d"
)
df.to_csv(csv_filename)
print(f"Data saved to {csv_filename}\n")
except Exception as e:
print(f"Failed to fetch from Tardis: {e}")
print("Falling back to loading cached data...")
df = pd.read_csv(csv_filename, index_col=0, parse_dates=True)
# Step 2: Set up Backtrader Cerebro engine
print("Step 2: Initializing Backtrader Cerebro engine...")
cerebro = bt.Cerebro(
cheat_on_open=True, # Use next day's open for trade execution
writer=True, # Enable CSV/console output
stdstats=True, # Show broker stats
live=False # Backtest mode (not live)
)
# Step 3: Add data feed
print("Step 3: Adding data feed...")
data_feed = PandasData(
dataname=df,
name=f"{exchange.upper()}:{symbol}"
)
cerebro.adddata(data_feed)
# Step 4: Configure broker
print("Step 4: Configuring broker...")
cerebro.broker.setcash(initial_cash)
cerebro.broker.setcommission(commission=0.001) # 0.1% taker fee (Binance spot)
# Step 5: Add strategy with custom parameters
print("Step 5: Adding AI Signal Strategy...")
cerebro.addstrategy(
AISignalStrategy,
ai_model=ai_model,
signal_interval=7, # Query AI every