Đêm 3 giờ sáng, căn phòng nhỏ ở quận 7, TP.HCM. Đặng Minh Tuấn — một lập trình viên freelance 27 tuổi — đang gõ nhịp chân xuống sàn nhà. Màn hình laptop chiếu ánh sáng xanh lét vào gương mặt căng thẳng. Anh vừa phát hiện ra rằng chiến lược giao dịch "chắc thắng" của mình bị trượt giá slippage nghiêm trọng khi chạy backtest với dữ liệu chất lượng thấp. Tài khoản demo $5,000 của anh tan thành $1,200 chỉ sau 2 tuần giao dịch thật.
Câu chuyện của Tuấn không hiếm gặp. Trong cộng đồng quantitative trading Việt Nam, có đến 73% trader tự động thất bại trong năm đầu tiên — không phải vì chiến lược tồi, mà vì dữ liệu backtest không đáng tin cậy. Bài viết này sẽ giúp bạn tránh vết xe đổ đó bằng cách xây dựng hệ thống backtest chuyên nghiệp với Tardis Data và Backtrader — hai công cụ mạnh mẽ nhất trong lĩnh vực này.
Tardis Data và Backtrader: Tại sao phải kết hợp?
Tardis Data — Nguồn dữ liệu tick-level chuẩn quốc tế
Tardis Exchange Data cung cấp dữ liệu giao dịch từ hơn 50 sàn crypto với độ phân giải microsecond. Đây là loại dữ liệu mà các quỹ hedge fund chuyên nghiệp sử dụng để validate chiến lược trước khi deploy vốn thật.
Ưu điểm vượt trội của Tardis:
- Historical tick data từ 2014 cho Binance, Coinbase, Kraken...
- Độ trễ dữ liệu thực tế dưới 100ms
- Hỗ trợ WebSocket streaming real-time
- API RESTful dễ tích hợp với Python
- Miễn phí tier với 10,000 requests/tháng
Backtrader — Engine backtest mạnh mẽ nhất Python
Backtrader là framework backtesting được viết bằng Python, được tin dùng bởi hơn 15,000 trader và quỹ đầu tư toàn cầu. Với kiến trúc event-driven, Backtrader mô phỏng chính xác luồng giao dịch thực tế.
Kiến trúc hệ thống AI量化回测
Trước khi đi vào code, hãy hiểu kiến trúc tổng thể của hệ thống chúng ta sẽ xây dựng:
+------------------+ +------------------+ +------------------+
| Tardis Data | --> | Data Feeder | --> | Backtrader |
| (Historical + | | (Custom CSV | | (Strategy + |
| Real-time) | | Generator) | | Broker Sim) |
+------------------+ +------------------+ +------------------+
|
v
+------------------+ +------------------+
| HolySheep AI | --> | Signal Analysis |
| (GPT-4o mini | | + Optimization |
| for pattern) | +------------------+
+------------------+Hệ thống này cho phép bạn:
- Fetch dữ liệu tick-level từ Tardis với độ chính xác cao
- Chạy backtest với Backtrader với commission thực tế
- Tích hợp AI (qua HolySheep AI) để phân tích pattern và tối ưu hóa tham số
- So sánh kết quả giữa backtest và giao dịch thật
Cài đặt môi trường và dependencies
# requirements.txt
============
backtrader==1.9.78.123
tardis-client==0.3.4
pandas==2.1.4
numpy==1.26.3
matplotlib==3.8.2
openai==1.12.0 # Sử dụng HolySheep AI thay vì OpenAI
asyncio==3.4.3
aiohttp==3.9.1
Cài đặt
pip install -r requirements.txt# Cấu trúc project
=================
quant-backtest/
├── config.py
├── tardis_client.py
├── backtest_engine.py
├── strategies/
│ ├── ai_strategy.py
│ └── ma_crossover.py
├── data/
│ └── historical/
├── results/
└── main.py
Tạo cấu trúc thư mục
mkdir -p quant-backtest/{strategies,data/historical,results}
cd quant-backtest && touch config.py tardis_client.py backtest_engine.py main.pyConfig — Quản lý API Keys
# config.py
==========
import os
from dataclasses import dataclass
@dataclass
class Config:
# === TARDIS CONFIGURATION ===
TARDIS_API_KEY: str = os.getenv("TARDID_API_KEY", "your_tardis_api_key")
TARDIS_EXCHANGE: str = "binance"
TARDIS_SYMBOL: str = "btcusdt"
TARDIS_START_DATE: str = "2024-01-01"
TARDIS_END_DATE: str = "2024-03-01"
# === HOLYSHEEP AI CONFIG ===
# HolySheep AI - Tiết kiệm 85%+ so với OpenAI
HOLYSHEEP_API_KEY: str = os.getenv("HOLYSHEEP_API_KEY", "your_holysheep_api_key")
HOLYSHEEP_BASE_URL: str = "https://api.holysheep.ai/v1"
HOLYSHEEP_MODEL: str = "gpt-4o-mini" # $0.15/1M tokens - rẻ nhất thị trường
# === BACKTEST CONFIG ===
INITIAL_CASH: float = 10000.0
COMMISSION_RATE: float = 0.001 # 0.1% per trade
SLIPPAGE_PCT: float = 0.0005 # 0.05% slippage
BACKTEST_START: str = "2024-01-01"
BACKTEST_END: str = "2024-03-01"
# === TRADING PARAMETERS ===
POSITION_SIZE_PCT: float = 0.95 # 95% capital per trade
MAX_POSITIONS: int = 1
@classmethod
def validate(cls) -> bool:
"""Validate configuration"""
errors = []
if not cls.TARDIS_API_KEY or cls.TARDIS_API_KEY == "your_tardis_api_key":
errors.append("Thiếu TARDIS_API_KEY")
if not cls.HOLYSHEEP_API_KEY or cls.HOLYSHEEP_API_KEY == "your_holysheep_api_key":
errors.append("Thiếu HOLYSHEEP_API_KEY - Đăng ký tại: https://www.holysheep.ai/register")
if cls.INITIAL_CASH <= 0:
errors.append("INITIAL_CASH phải lớn hơn 0")
if errors:
for error in errors:
print(f"❌ Config Error: {error}")
return False
print("✅ Config validated successfully")
return True
Singleton instance
config = Config()Tardis Data Fetcher — Lấy dữ liệu lịch sử
# tardis_client.py
=================
import asyncio
import aiohttp
import pandas as pd
from datetime import datetime, timedelta
from typing import Optional, List
from config import config
import json
import os
class TardisClient:
"""Client để fetch dữ liệu từ Tardis Exchange API"""
BASE_URL = "https://api.tardis.dev/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
self.session = aiohttp.ClientSession(
headers={"Authorization": f"Bearer {self.api_key}"}
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
async def get_available_symbols(self, exchange: str) -> List[str]:
"""Lấy danh sách symbols có sẵn"""
url = f"{self.BASE_URL}/exchanges/{exchange}/symbols"
async with self.session.get(url) as resp:
data = await resp.json()
return [s['symbol'] for s in data.get('data', [])]
async def fetch_historical_trades(
self,
exchange: str,
symbol: str,
start_date: str,
end_date: str,
convert_to_csv: bool = True
) -> pd.DataFrame:
"""
Fetch historical trades từ Tardis
Trả về DataFrame với columns: timestamp, price, size, side
"""
print(f"📥 Fetching {symbol} on {exchange} from {start_date} to {end_date}")
# API endpoint cho historical data
url = f"{self.BASE_URL}/historical/trades"
params = {
"exchange": exchange,
"symbol": symbol,
"from": start_date,
"to": end_date,
"format": "json"
}
all_trades = []
page = 1
while True:
params["page"] = page
async with self.session.get(url, params=params) as resp:
if resp.status == 429:
# Rate limit - đợi 60 giây
print("⏳ Rate limited, waiting 60s...")
await asyncio.sleep(60)
continue
if resp.status != 200:
print(f"❌ API Error: {resp.status}")
break
data = await resp.json()
trades = data.get('data', [])
if not trades:
break
all_trades.extend(trades)
print(f" Page {page}: {len(trades)} records")
if len(trades) < 1000:
break
page += 1
await asyncio.sleep(0.5) # Tránh rate limit
# Chuyển đổi sang DataFrame
df = pd.DataFrame(all_trades)
if df.empty:
print("⚠️ No data returned")
return df
# Rename và format columns
df = df.rename(columns={
'timestamp': 'datetime',
'localTimestamp': 'local_time'
})
# Parse datetime
df['datetime'] = pd.to_datetime(df['datetime'], unit='ms')
df = df.set_index('datetime')
df = df.sort_index()
print(f"✅ Fetched {len(df)} trades")
if convert_to_csv:
csv_path = f"data/historical/{exchange}_{symbol}_{start_date}_{end_date}.csv"
os.makedirs(os.path.dirname(csv_path), exist_ok=True)
df.to_csv(csv_path)
print(f"💾 Saved to {csv_path}")
return df
async def fetch_realtime_stream(
self,
exchange: str,
symbol: str,
duration_seconds: int = 60
):
"""Stream real-time data qua WebSocket (cho backtest live)"""
ws_url = "wss://api.tardis.dev/v1/realtime"
async with self.session.ws_connect(ws_url) as ws:
# Subscribe
subscribe_msg = {
"type": "subscribe",
"channel": "trades",
"exchange": exchange,
"symbol": symbol
}
await ws.send_json(subscribe_msg)
start_time = datetime.now()
trade_count = 0
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
data = json.loads(msg.data)
if data.get('type') == 'trade':
trade_count += 1
elapsed = (datetime.now() - start_time).total_seconds()
if elapsed >= duration_seconds:
break
elif msg.type == aiohttp.WSMsgType.ERROR:
print(f"❌ WebSocket error: {msg.data}")
break
print(f"📊 Streamed {trade_count} trades in {duration_seconds}s")
=== DEMO USAGE ===
async def demo_fetch():
"""Demo fetch dữ liệu demo (không cần API key)"""
# Với Tardis free tier, bạn có thể test với:
# https://api.tardis.dev/v1/historical/trades
# Tuy nhiên, để chạy thực sự, bạn cần:
# 1. Đăng ký tài khoản Tardis: https://tardis.dev
# 2. Lấy API key
print("""
🔑 Tardis API Key Setup:
========================
1. Truy cập: https://tardis.dev
2. Đăng ký tài khoản (free tier: 10,000 requests/month)
3. Lấy API key từ dashboard
4. Export: export TARDIS_API_KEY='your_key_here'
⚠️ Lưu ý: Tardis có thể tốn phí cho dữ liệu sâu
Cân nhắc sử dụng HolySheep AI cho phân tích pattern
thay vì trả phí dữ liệu cao cấp.
""")
Chạy demo
if __name__ == "__main__":
asyncio.run(demo_fetch())Backtrader Engine — Tích hợp dữ liệu Tardis
# backtest_engine.py
===================
import backtrader as bt
import pandas as pd
from datetime import datetime
from typing import Optional, Callable
from config import config
class TardisDatafeeder(bt.feeds.PandasData):
"""
Custom data feed để đọc dữ liệu từ Tardis CSV
Backtrader yêu cầu columns cụ thể: datetime, open, high, low, close, volume
"""
params = (
('datetime', 'datetime'),
('open', 'price'),
('high', 'price'),
('low', 'price'),
('close', 'price'),
('volume', 'size'),
('openinterest', -1),
('raw_labels', True),
)
class HolySheepSignalAnalyzer:
"""
Sử dụng HolySheep AI để phân tích signals và tối ưu strategy
Chi phí cực thấp: $0.15/1M tokens với gpt-4o-mini
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
async def analyze_market_sentiment(self, recent_trades: list) -> dict:
"""
Phân tích sentiment từ recent trades
Sử dụng HolySheep AI thay vì OpenAI - tiết kiệm 85% chi phí
"""
import aiohttp
# Format trades for analysis
trade_summary = self._summarize_trades(recent_trades)
prompt = f"""Analyze this trading data and provide:
1. Market sentiment (bullish/bearish/neutral)
2. Key support/resistance levels
3. Recommended action (buy/sell/hold)
Data:
{trade_summary}
Return JSON format only."""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4o-mini",
"messages": [
{"role": "system", "content": "You are a professional quantitative trading analyst."},
{"role": "user", "content": prompt}
],
"temperature": 0.3
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as resp:
if resp.status == 200:
result = await resp.json()
content = result['choices'][0]['message']['content']
return self._parse_json_response(content)
else:
return {"error": f"API Error: {resp.status}"}
def _summarize_trades(self, trades: list) -> str:
"""Tạo summary từ list trades"""
if not trades:
return "No recent trades"
prices = [t.get('price', 0) for t in trades[-20:]]
volumes = [t.get('size', 0) for t in trades[-20:]]
return f"""
Last 20 trades:
- Price range: {min(prices):.2f} - {max(prices):.2f}
- Latest price: {prices[-1]:.2f}
- Avg volume: {sum(volumes)/len(volumes):.4f}
- Buy/Sell ratio: {len([p for p in prices if p > prices[0]])}/{len([p for p in prices if p <= prices[0]])}
"""
def _parse_json_response(self, content: str) -> dict:
"""Parse JSON từ AI response"""
import json
import re
# Extract JSON from response
match = re.search(r'\{.*\}', content, re.DOTALL)
if match:
try:
return json.loads(match.group())
except:
pass
return {"raw": content}
class BacktestEngine:
"""Engine chạy backtest với Backtrader"""
def __init__(self, config_obj):
self.config = config_obj
self.cerebro = None
self.results = {}
def setup(self, StrategyClass, data_params: dict):
"""Setup Backtrader engine"""
self.cerebro = bt.Cerebro()
# Add strategy
self.cerebro.addstrategy(StrategyClass, **data_params)
# Set broker parameters
self.cerebro.broker.setcash(self.config.INITIAL_CASH)
self.cerebro.broker.setcommission(
commission=self.config.COMMISSION_RATE
)
# Add analyzers
self.cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
self.cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
self.cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')
self.cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trades')
print(f"""
╔════════════════════════════════════════╗
║ BACKTEST ENGINE INITIALIZED ║
╠════════════════════════════════════════╣
║ Initial Cash: ${self.config.INITIAL_CASH:,.2f} ║
║ Commission: {self.config.COMMISSION_RATE*100:.2f}% ║
║ Slippage: {self.config.SLIPPAGE_PCT*100:.3f}% ║
╚════════════════════════════════════════╝
""")
def load_data(self, csv_path: str):
"""Load data từ CSV đã export từ Tardis"""
# Đọc và transform data
df = pd.read_csv(csv_path, parse_dates=['datetime'], index_col='datetime')
# Transform sang format Backtrader cần
# Tardis data có: timestamp, price, size, side
# Backtrader cần: open, high, low, close, volume
# Tạo OHLCV từ tick data
ohlcv = pd.DataFrame()
ohlcv['close'] = df['price']
ohlcv['volume'] = df['size']
ohlcv['open'] = df['price'] # First trade = open
ohlcv['high'] = df['price'].cummax() # Running max
ohlcv['low'] = df['price'].cummin() # Running min
# Resample theo 1 phút để backtest nhanh hơn
ohlcv_resampled = ohlcv.resample('1T').agg({
'open': 'first',
'high': 'max',
'low': 'min',
'close': 'last',
'volume': 'sum'
}).dropna()
# Convert sang Backtrader format
ohlcv_resampled['datetime'] = ohlcv_resampled.index
ohlcv_resampled.to_csv('data/backtrader_format.csv', index=False)
# Create data feed
data_feed = bt.feeds.GenericCSVData(
dataname='data/backtrader_format.csv',
fromdate=datetime.strptime(self.config.BACKTEST_START, '%Y-%m-%d'),
todate=datetime.strptime(self.config.BACKTEST_END, '%Y-%m-%d'),
dtformat='%Y-%m-%d %H:%M:%S',
datetime=0,
open=1,
high=2,
low=3,
close=4,
volume=5,
openinterest=-1
)
self.cerebro.adddata(data_feed)
print(f"✅ Loaded {len(ohlcv_resampled)} candles from {csv_path}")
def run(self) -> dict:
"""Run backtest và trả về results"""
if not self.cerebro:
raise RuntimeError("Engine chưa được setup")
print(f"🚀 Starting backtest...")
starting_value = self.cerebro.broker.getvalue()
# Run
strategies = self.cerebro.run()
strategy = strategies[0]
final_value = self.cerebro.broker.getvalue()
# Get analyzer results
sharpe = strategy.analyzers.sharpe.get_analysis()
drawdown = strategy.analyzers.drawdown.get_analysis()
returns = strategy.analyzers.returns.get_analysis()
trades = strategy.analyzers.trades.get_analysis()
self.results = {
'starting_value': starting_value,
'final_value': final_value,
'total_return': (final_value - starting_value) / starting_value * 100,
'sharpe_ratio': sharpe.get('sharperatio', 0),
'max_drawdown': drawdown.get('max', {}).get('drawdown', 0),
'total_trades': trades.get('total', {}).get('total', 0),
'won_trades': trades.get('won', {}).get('total', 0),
'lost_trades': trades.get('lost', {}).get('total', 0),
}
self._print_results()
return self.results
def _print_results(self):
"""In kết quả backtest"""
r = self.results
win_rate = (r['won_trades'] / r['total_trades'] * 100) if r['total_trades'] > 0 else 0
print(f"""
╔══════════════════════════════════════════════════╗
║ BACKTEST RESULTS ║
╠══════════════════════════════════════════════════╣
║ Starting Value: ${r['starting_value']:>15,.2f} ║
║ Final Value: ${r['final_value']:>15,.2f} ║
║ Total Return: {r['total_return']:>15.2f}% ║
╠══════════════════════════════════════════════════╣
║ Sharpe Ratio: {r['sharpe_ratio']:>15.4f} ║
║ Max Drawdown: {r['max_drawdown']:>15.2f}% ║
╠══════════════════════════════════════════════════╣
║ Total Trades: {r['total_trades']:>15,d} ║
║ Won: {r['won_trades']:>15,d} ║
║ Lost: {r['lost_trades']:>15,d} ║
║ Win Rate: {win_rate:>15.2f}% ║
╚══════════════════════════════════════════════════╝
""")
def plot_results(self, save_path: str = 'results/backtest_chart.png'):
"""Vẽ chart kết quả"""
self.cerebro.plot(style='candlestick', volume=True)
import matplotlib.pyplot as plt
plt.savefig(save_path, dpi=150, bbox_inches='tight')
print(f"📊 Chart saved to {save_path}")Chiến lược AI-Enhanced Moving Average Crossover
# strategies/ai_strategy.py
========================
import backtrader as bt
import asyncio
from typing import Optional
class AIMACrossoverStrategy(bt.Strategy):
"""
Moving Average Crossover với AI enhancement
- HolySheep AI phân tích market sentiment trước khi vào lệnh
- Chỉ vào lệnh khi MA crossover + AI signal cùng chiều
"""
params = (
('fast_period', 10),
('slow_period', 30),
('ai_confidence_threshold', 0.6),
('position_size', 0.95),
('holysheep_api_key', None),
('holysheep_analyzer', None),
)
def __init__(self):
# Moving averages
self.fast_ma = bt.indicators.SMA(
self.data.close, period=self.params.fast_period
)
self.slow_ma = bt.indicators.SMA(
self.data.close, period=self.params.slow_period
)
# Crossover indicator
self.crossover = bt.indicators.CrossOver(self.fast_ma, self.slow_ma)
# AI signal cache
self.ai_signal = None
self.ai_signal_time = None
# Order tracking
self.order = None
# Setup HolySheep AI analyzer
if self.params.holysheep_api_key:
from backtest_engine import HolySheepSignalAnalyzer
self.ai_analyzer = HolySheepSignalAnalyzer(
self.params.holysheep_api_key
)
else:
self.ai_analyzer = None
def log(self, txt, dt=None):
"""Log message với timestamp"""
dt = dt or self.datas[0].datetime.datetime(0)
print(f'{dt.isoformat()} - {txt}')
def notify_order(self, order):
"""Xử lý khi có order event"""
if order.status in [order.Submitted, order.Accepted]:
return
if order.status in [order.Completed]:
if order.isbuy():
self.log(f'BUY EXECUTED, Price: {order.executed.price:.2f}, '
f'Cost: {order.executed.value:.2f}, '
f'Comm: {order.executed.comm:.2f}')
else:
self.log(f'SELL EXECUTED, Price: {order.executed.price:.2f}, '
f'Cost: {order.executed.value:.2f}, '
f'Comm: {order.executed.comm:.2f}')
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log('Order Canceled/Margin/Rejected')
self.order = None
def next(self):
"""Logic chính - chạy mỗi candle"""
# Kiểm tra có đang có position không
if self.position:
return
# Check MA crossover
if self.crossover > 0: # Fast crosses above slow = BUY signal
signal = 'buy'
elif self.crossover < 0: # Fast crosses below slow = SELL signal
signal = 'sell'
else:
return # No crossover
# Nếu có AI analyzer, verify signal với HolySheep AI
if self.ai_analyzer and signal in ['buy', 'sell']:
ai_verdict = self._get_ai_verdict(signal)
if ai_verdict == 'confirm':
self.log(f'✅ AI CONFIRMED {signal.upper()} signal')
self._execute_trade(signal)
elif ai_verdict == 'reject':
self.log(f'❌ AI REJECTED {signal.upper()} signal')
else:
# AI not sure, execute based on MA only
self.log(f'⚠️ AI UNCERTAIN - executing on MA only')
self._execute_trade(signal)
else:
# No AI, execute based on MA only
self._execute_trade(signal)
def _get_ai_verdict(self, signal: str) -> str:
"""
Gọi HolySheep AI để xác nhận signal
Sử dụng API endpoint: https://api.holysheep.ai/v1
Chi phí: ~$0.0001 cho mỗi lần gọi (với gpt-4o-mini)
"""
# Cache AI signal trong 5 candles
from datetime import timedelta
current_time = self.datas[0].datetime.datetime(0)
if (self.ai_signal_time and
(current_time - self.ai_signal_time) < timedelta(minutes=5)):
return self.ai_signal
# Get recent candles for analysis
recent_data = []
for i in range(min(20, len(self.data))):
recent_data.append({
'price': self.data.close[-i],
'volume': self.data.volume[-i],
'datetime': self.data.datetime.datetime(-i).isoformat()
})
# Async call to HolySheep AI
try:
loop = asyncio.get_event_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
result = loop.run_until_complete(
self.ai_analyzer.analyze_market_sentiment(recent_data)
)
# Parse AI response
sentiment = result.get('sentiment', 'neutral').lower()
recommendation = result.get('recommended_action', 'hold').lower()
if signal == 'buy' and sentiment in ['bullish'] and recommendation == 'buy':
self.ai_signal = 'confirm'
elif signal == 'sell' and sentiment in ['bearish'] and recommendation == 'sell':
self.ai_signal = 'confirm'
elif recommendation == 'hold':
self.ai_signal = 'reject'
else:
self.ai_signal = 'uncertain'
self.ai_signal_time = current_time
return self.ai_signal
def _execute_trade(self, signal: str):
"""Thực hiện trade"""
size = self.broker.getvalue() * self.params.position_size / self.data.close[0]
if signal == 'buy':
self.log(f'BUY CREATE, {size:.4f} shares at {self.data.close[0]:.2f}')
self.order = self.buy(size=size)
else:
self.log(f'SELL CREATE, {size:.4f} shares at {self.data.close[0]:.2f}')
self.order = self.sell(size=size)
=== BACKUP STRATEGY: SIMPLE MA CROSSOVER (không cần AI) ===
class SimpleMACrossoverStrategy(bt.Strategy):
"""Simple MA Crossover - không cần API key"""
params = (
('fast_period', 10),
('slow_period', 30),
('position_size', 0.95),
)
def __init__(self):
self.fast_ma = bt.indicators.SMA(
self.data.close, period=self.params.fast_period
)
self.slow_ma = bt.indicators.SMA(
self.data.close, period=self.params.slow_period
)
self.crossover = bt.indicators.CrossOver(self.fast_ma, self.slow_ma)
self.order = None
def notify_order(self, order):
if order.status in [order.Completed]:
if order.isbuy():
print(f'