Backtester works
This commit is contained in:
+30137
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,27 @@
|
||||
print("File is running")
|
||||
import pandas as pd
|
||||
from data.model import Candle
|
||||
|
||||
def load_candles(filepath: str) -> list[Candle]:
|
||||
df = pd.read_csv(filepath, sep=";", header=None, names=["timestamp", "open", "high", "low", "close", "volume"])
|
||||
df["timestamp"] = pd.to_datetime(df["timestamp"], format="%Y%m%d %H%M%S")
|
||||
|
||||
candles = []
|
||||
for _, row in df.iterrows():
|
||||
candle = Candle(
|
||||
time_open=row["timestamp"],
|
||||
open=row["open"],
|
||||
high=row["high"],
|
||||
low=row["low"],
|
||||
close=row["close"],
|
||||
volume=row["volume"]
|
||||
)
|
||||
candles.append(candle)
|
||||
|
||||
return candles
|
||||
|
||||
if __name__ == "__main__":
|
||||
candles = load_candles("data.csv")
|
||||
print(f"Loaded {len(candles)} candles")
|
||||
print(f"First: {candles[0]}")
|
||||
print(f"Last: {candles[-1]}")
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
|
||||
@dataclass
|
||||
class Candle:
|
||||
|
||||
@@ -0,0 +1,37 @@
|
||||
from data.model import Candle, Trade
|
||||
from strategies.base import SimpleStrategy
|
||||
|
||||
# takes a list of Candles and a starting balance, and returns a list of Trades
|
||||
def run_backtest(candles: list[Candle], starting_balance: float) -> list[Trade]:
|
||||
balance = starting_balance
|
||||
trades = []
|
||||
position = None
|
||||
strategy = SimpleStrategy()
|
||||
|
||||
for i, candle in enumerate(candles):
|
||||
# pass 'i' or the sliced history to the strategy
|
||||
signal = strategy.check_signal(candles[:i+1])
|
||||
# If signal and no position, open trade
|
||||
if signal == "BUY" and position is None:
|
||||
position = {
|
||||
"type": "long",
|
||||
"entry_price": candle.close,
|
||||
"enter_time": candle.time_open
|
||||
}
|
||||
|
||||
# If signal and in position, close trade
|
||||
elif signal == "SELL" and position is not None:
|
||||
trade = Trade(
|
||||
enter_time=position["enter_time"],
|
||||
enter_price=position["entry_price"],
|
||||
direction="long",
|
||||
exit_time=candle.time_open,
|
||||
exit_price=candle.close,
|
||||
pnl=candle.close - position["entry_price"]
|
||||
)
|
||||
trades.append(trade)
|
||||
position = None
|
||||
|
||||
|
||||
|
||||
return trades
|
||||
@@ -0,0 +1,11 @@
|
||||
from data.loader import load_candles
|
||||
from engine.backtester import run_backtest
|
||||
|
||||
candles = load_candles("data/data.csv")
|
||||
print(f"Loaded {len(candles)} candles")
|
||||
|
||||
trades = run_backtest(candles, starting_balance=10000.0)
|
||||
print(f"Completed {len(trades)} trades")
|
||||
|
||||
for trade in trades[:5]:
|
||||
print(trade)
|
||||
@@ -0,0 +1,21 @@
|
||||
class SimpleStrategy:
|
||||
def __init__(self, fast_period=10, slow_period=20):
|
||||
self.fast_period = fast_period
|
||||
self.slow_period = slow_period
|
||||
|
||||
def check_signal(self, history):
|
||||
if len(history) < self.slow_period + 1:
|
||||
return None
|
||||
|
||||
fast_now = sum(c.close for c in history[-self.fast_period:]) / self.fast_period
|
||||
slow_now = sum(c.close for c in history[-self.slow_period:]) / self.slow_period
|
||||
|
||||
fast_prev = sum(c.close for c in history[-self.fast_period-1:-1]) / self.fast_period
|
||||
slow_prev = sum(c.close for c in history[-self.slow_period-1:-1]) / self.slow_period
|
||||
|
||||
if fast_prev <= slow_prev and fast_now > slow_now:
|
||||
return "BUY"
|
||||
elif fast_prev >= slow_prev and fast_now < slow_now:
|
||||
return "SELL"
|
||||
|
||||
return None
|
||||
Reference in New Issue
Block a user