Does not execute trades

[Bl4ckVo1d]

Tried several examples, trades are not made at all.
Switching to FreePascal and will write my own code THAT WILL WORK.

import pandas as pd
import numpy as np
import talib
import backtrader as bt
from datetime import datetime

def load_data(file_path, symbol="BTCUSDT"):
"""Load OHLCV data from CSV, filter by symbol, and prepare for backtesting."""
# Read raw file content
with open(file_path, "r", encoding="utf-8") as f:
raw_lines = [next(f).strip() for _ in range(2)]
print("Raw First 2 Lines:", raw_lines)
# Load CSV without parse_dates initially
df = pd.read_csv(
file_path,
sep=",",
decimal=".",
thousands=None,
encoding="utf-8",
engine="python"
)
print("CSV Columns:", df.columns.tolist())
# Clean column names
df.columns = df.columns.str.strip().str.replace('"', '')
print("Cleaned CSV Columns:", df.columns.tolist())
# Manually parse Date
if "Date" in df.columns:
df["Date"] = pd.to_datetime(df["Date"], format="mixed", errors="coerce")
df = df.dropna(subset=["Date"]) # Drop rows with unparseable dates
df.set_index("Date", inplace=True)
print(df.index)
else:
raise ValueError("Date column not found after cleaning")
# Ensure index is DatetimeIndex
if not isinstance(df.index, pd.DatetimeIndex):
df.index = pd.to_datetime(df.index, format="mixed", errors="coerce")
# Filter and format
df = df[df["Symbol"] == symbol]
df = df[["Open", "High", "Low", "Close", "Volume BTC"]]
df.columns = ["open", "high", "low", "close", "volume"]
return df

def calculate_indicators(df):
"""Calculate common technical indicators using ta-lib."""
indicators = {}
indicators["sma20"] = talib.SMA(df["close"], timeperiod=20)
indicators["rsi14"] = talib.RSI(df["close"], timeperiod=14)
return indicators

class CustomCSVData(bt.feeds.PandasData):
"""Custom Backtrader data feed for CSV data."""
params = (
("datetime", None), # Use index
("open", 0),
("high", 1),
("low", 2),
("close", 3),
("volume", 4),
("openinterest", -1),
)

def run_backtest(data, strategy, cash=1000000):
"""Run a backtest with given data and strategy."""
cerebro = bt.Cerebro()
cerebro.adddata(data)
cerebro.addstrategy(strategy)
cerebro.broker.setcash(cash)
cerebro.broker.setcommission(commission=0.001) # 0.1% fee per trade
cerebro.addsizer(bt.sizers.FixedSize, stake=1) # Ensure trade size is applied

# Add observers for diagnostics
cerebro.addobserver(bt.observers.Broker)
cerebro.addobserver(bt.observers.Value)
cerebro.addobserver(bt.observers.Trades)

# Print starting portfolio value
print('Starting Portfolio Value: %.2f' % cerebro.broker.getvalue())

# Run the backtest with runonce=False for more accurate simulation
cerebro.run(runonce=False)

# Print final portfolio value
print('Final Portfolio Value: %.2f' % cerebro.broker.getvalue())
return cerebro.broker.getvalue()

Example strategy (SMA Crossover)

class SMAStrategy(bt.Strategy):
params = (("sma_period", 20), ("max_hold_period", 5))

def __init__(self):
    self.sma = bt.indicators.SMA(self.data.close, period=self.params.sma_period)
    self.order = None
    self.bars_since_entry = 0
    self.in_position = False

def next(self):
    if self.order:
        return

    print(f"{self.data.datetime.datetime(0)}: Close {self.data.close[0]}, SMA {self.sma[0]}")

    if not self.in_position:
        if self.data.close[0] > self.sma[0]:
            print(f"Submitting BUY @ {self.data.close[0]} on {self.data.datetime.datetime(0)}")
            self.order = self.buy()
            self.bars_since_entry = 0
    else:
        self.bars_since_entry += 1
        if self.bars_since_entry >= self.params.max_hold_period:
            print(f"Submitting SELL @ {self.data.close[0]} on {self.data.datetime.datetime(0)} (Max hold period reached: {self.bars_since_entry} bars)")
            self.order = self.sell()
        else:
            print(f"No sell at {self.data.datetime.datetime(0)}: Bars since entry: {self.bars_since_entry}")

def notify_order(self, order):
    if order.status in [order.Submitted, order.Accepted]:
        return  # Wait for completion
    if order.status in [order.Completed]:
        if order.isbuy():
            print(f"BUY EXECUTED @ {order.executed.price} on {self.data.datetime.datetime(0)}")
            self.in_position = True
        elif order.issell():
            print(f"SELL EXECUTED @ {order.executed.price} on {self.data.datetime.datetime(0)}")
            self.in_position = False
    elif order.status in [order.Canceled, order.Margin, order.Rejected]:
        print("ORDER FAILED: ", order.status)
        self.in_position = False
    self.order = None

class TestStrategy(bt.Strategy):

def log(self, txt, dt=None):
    ''' Logging function fot this strategy'''
    dt = dt or self.datas[0].datetime.date(0)
    print('%s, %s' % (dt.isoformat(), txt))

def __init__(self):
    # Keep a reference to the "close" line in the data[0] dataseries
    self.dataclose = self.datas[0].close

    # To keep track of pending orders
    self.order = None

def notify_order(self, order):
    if order.status in [order.Submitted, order.Accepted]:
        # Buy/Sell order submitted/accepted to/by broker - Nothing to do
        return

    # Check if an order has been completed
    # Attention: broker could reject order if not enough cash
    if order.status in [order.Completed]:
        if order.isbuy():
            self.log('BUY EXECUTED, %.2f' % order.executed.price)
        elif order.issell():
            self.log('SELL EXECUTED, %.2f' % order.executed.price)

        self.bar_executed = len(self)

    elif order.status in [order.Canceled, order.Margin, order.Rejected]:
        self.log('Order Canceled/Margin/Rejected')

    # Write down: no pending order
    self.order = None

def next(self):
    # Simply log the closing price of the series from the reference
    self.log('Close, %.2f' % self.dataclose[0])

    # Check if an order is pending ... if yes, we cannot send a 2nd one
    if self.order:
        return

    # Check if we are in the market
    if not self.position:

        # Not yet ... we MIGHT BUY if ...
        if self.dataclose[0] < self.dataclose[-1]:
                # current close less than previous close

                if self.dataclose[-1] < self.dataclose[-2]:
                    # previous close less than the previous close

                    # BUY, BUY, BUY!!! (with default parameters)
                    self.log('BUY CREATE, %.2f' % self.dataclose[0])

                    # Keep track of the created order to avoid a 2nd order
                    self.order = self.buy()

    else:

        # Already in the market ... we might sell
        if len(self) >= (self.bar_executed + 5):
            # SELL, SELL, SELL!!! (with all possible default parameters)
            self.log('SELL CREATE, %.2f' % self.dataclose[0])

            # Keep track of the created order to avoid a 2nd order
            self.order = self.sell()

Test data loading

if name == "main":
file_path = r"C:/Users/bvarh/OneDrive/BackTesting/Data/Binance_BTCUSDT_1h.csv"
print("Starting data load...")
df = load_data(file_path)
df = df.tail(1000) # Use last 1000 rows (approx. 41 days of 1h data)
print("Data Head:\n", df.head())
print("Data Rows:", len(df))
print("Calculating indicators...")
indicators = calculate_indicators(df)
print("First 5 SMA20:\n", indicators["sma20"][:5])
print("Setting up backtest...")
data_feed = CustomCSVData(dataname=df)
print("Running backtest...")
final_value = run_backtest(data_feed, TestStrategy)
print("Final Portfolio Value:", final_value)