-
Notifications
You must be signed in to change notification settings - Fork 22
Loading OHLC data from a custom source
By default, Stock-pattern uses data from CSV files. It uses the Loader classes EODFileLoader and IEODFileLoader which are defined in the src/loaders
folder.
The loader class implements the AbstractLoader
class. It has just two methods get
and close
, and three properties. These must be defined and implemented by all loader classes.
timeframes: This is a dictionary of key-value pairs. The key is any value passed to the --tf
CLI option. The value can be either offset strings used in pandas.DataFrame.resample or timeframe string passed to the web API.
timeframes = {"daily": "D", "weekly": "W-SUN", "monthly": "MS"}
tf: It is the currently active timeframe. It can be set during initialization, since it never changes during the script's lifetime.
closed: It indicates the current status of the loader. It is False on initialization. If True, the close
method is never called.
AbstractLoader class defines two methods. You're free to define additional helper methods as required.
get: This method takes a single parameter the symbol name. It returns the Pandas Dataframe. It may return None in case of any errors.
close: This method performs cleanup actions like closing a Database or TCP connection. The closed
property must be set to True when called.
We can create a custom Loader to load data from the Bybit API using their official library Pybit.
pip install pybit pycryptodome
Create a file named ByBitLoader.py
in src/loaders
. Import AbstractLoader and create the ByBitLoader
class implementing AbstractLoader.
Pointers to the below code:
- I refered the ByBit docs and added the timeframes dictionary.
- I initialize the pybit class while initializing ByBitLoader. Do not initialize or establish database connections within the
get
method. - The
get
method calls the pybit's get_kline method passing the necessary parameters. - The
close
method is not implemented since there is nothing to close. I have set theclosed
property to True, soclose
is never called.
from .AbstractLoader import AbstractLoader
from pybit.unified_trading import HTTP
from typing import Optional
from datetime import datetime
import pandas as pd
class ByBitLoader(AbstractLoader):
timeframes = {
"1": 1,
"5": 5,
"15": 15,
"30": 30,
"60": 60,
"2h": 120,
"4h": 240,
"daily": "D",
}
def __init__(
self,
config: dict,
tf: Optional[str] = None,
end_date: Optional[datetime] = None,
period: int = 160,
):
self.closed = True
self.period = period
self.tf = config.get("DEFAULT_TF", "daily") if tf is None else tf
self.interval = self.timeframes[self.tf]
# Initialise Pybit
self.session = HTTP(testnet=True)
# Define columns for DataFrame
self.columns = (
"Date",
"Open",
"High",
"Low",
"Close",
"Volume",
"Turnover",
)
def get(self, symbol: str) -> Optional[pd.DataFrame]:
result = self.session.get_kline(
symbol=symbol.upper(), interval=self.interval, limit=self.period
)
df = pd.DataFrame(
result["result"]["list"], columns=self.columns, dtype="float32"
)
df["Date"] = pd.to_datetime(df["Date"], unit="ms")
df.set_index("Date", inplace=True)
return df.iloc[::-1]
def close(self):
"""Not required here as nothing to close"""
pass
Once you have completed writing the Loader class, add it to your user.json as below.
{
"DATA_PATH": "~/Documents/python/eod2/src/eod2_data/daily",
"SYM_LIST": "~/Documents/python/stock-pattern/src/data.csv",
"POST_SCAN_PLOT": true,
"LOADER": "ByBitLoader"
}
Finally, run init.py
py init.py --sym btcusdt --tf 60 -p all
The above code is a minimal and barebones implementation.
- There is no error handling,
- No API throttling. Stock-pattern uses multiple processes to execute, which could exceed ByBits API limits. Use a library like ratelimit to implement throttling.
- No caching mechanism. The
get
will get called twice, once during the scan and when plotting the charts. It means two network requests.