"""
This is the main classes file.
"""
from .version import __version__
import os
import pandas as pd
from random import choice
from time import time
import numpy as np
from .exchange import Exchange
from .auxiliar import csv_klines_setup, check_continuity, repair_kline_discontinuity
from .core.exceptions import BinPanException
from .core.secrets import get_secret
from .core.trades import Trades
from .api.exchange_info import (get_decimal_positions, get_info_dic, get_precision, get_orderTypes_and_permissions, get_fees,
get_symbols_filters)
from .storage.files import select_file, read_csv_to_dataframe, save_dataframe_to_csv, extract_filename_metadata
from .analysis.indicators import reversal_candles
from .core.logs import LogManager
from .api.market import (get_candles_by_time_stamps, parse_candles_to_dataframe, convert_to_numeric, basic_dataframe,
get_historical_agg_trades, parse_agg_trades_to_dataframe, get_historical_atomic_trades,
parse_atomic_trades_to_dataframe, get_order_book)
# plotting.charts se importa lazy para evitar cargar plotly+scipy al importar binpan
def _plotting():
from binpan.plotting import charts
return charts
from .core.time_helper import (check_tick_interval, convert_milliseconds_to_str, tick_interval_values)
# analysis.aggregations se importa lazy para evitar cargar scipy al importar binpan
from .core.standards import (binance_api_candles_cols, agg_trades_api_map_columns, atomic_trades_api_map_columns, time_cols,
dts_time_cols, reversal_columns_order, agg_trades_columns_from_binance, atomic_trades_columns_from_binance)
from .api.wallet_api import convert_str_date_to_ms
import warnings
# analysis.numba_tools se importa lazy para evitar cargar numba al importar binpan
from .core.timeframes import Timeframe
binpan_logger = LogManager(filename='./logs/binpan.log', name='binpan', info_level='INFO')
version = __version__
empty_agg_trades_msg = "Empty trades, please request using: get_agg_trades() method: Example: my_symbol.get_agg_trades()"
empty_atomic_trades_msg = "Empty atomic trades, please request using: get_atomic_trades() method: Example: my_symbol.get_atomic_trades()"
from .symbol_indicators import SymbolIndicators
from .symbol_plotting import SymbolPlotting
from .symbol_strategy import SymbolStrategy
[docs]
class Symbol(SymbolIndicators, SymbolPlotting, SymbolStrategy):
"""
Creates an object from binance klines and/or trade data. It contains the raw api response, a dataframe that can be modified and many
more.
Symbols can be any trading pair on Binance, such as BTCUSDT or ETHBUSD, and time intervals can be specified using the
following strings:
```
'1m', '3m', '5m', '15m', '30m', '1h', '2h', '4h', '6h', '8h', '12h', '1d', '3d', '1w', or '1M'
```
The class provides several plotting methods for quick data visualization.
:param str symbol: It can be any symbol in the binance exchange, like BTCUSDT, ethbusd or any other. Capital letters don't matter.
:param str tick_interval: Any candle's interval available in binance. Capital letters don't matter.
:param int or str start_time: It can be an integer in milliseconds from epoch (1970-01-01 00:00:00 UTC) or any string in the formats:
.. code-block::
- %Y-%m-%d %H:%M:%S.%f: **2022-05-11 06:45:42.124567**
- %Y-%m-%d %H:%M:%S: **2022-05-11 06:45:42**
If start time is passed, it gets the next open according to the tick interval selected except an exact open time passed.
:param int or str end_time: It can be an integer in milliseconds from epoch (1970-01-01 00:00:00 UTC) or any string in the formats:
.. code-block::
- %Y-%m-%d %H:%M:%S.%f: **2022-05-11 06:45:42.124**
- %Y-%m-%d %H:%M:%S: **2022-05-11 06:45:42**
If end time is passed, it gets candles till the previous close according to the selected tick interval.
Example with daily time saving inside the interval:
.. code-block:: python
import binpan
btcusdt = binpan.Symbol(symbol='btcusdt',
tick_interval='15m',
time_zone='Europe/Madrid',
start_time='2021-10-31 01:00:00',
end_time='2021-10-31 03:00:00')
btcusdt.df
Open High Low Close Volume
BTCUSDT 15m Europe/Madrid
2021-10-31 01:00:00+02:00 61540.32 61780.44 61488.35 61737.39 316.93504
2021-10-31 01:15:00+02:00 61737.39 61748.13 61486.86 61637.99 485.50681
2021-10-31 01:30:00+02:00 61637.99 61777.71 61479.76 61743.32 441.29718
2021-10-31 01:45:00+02:00 61743.31 61900.00 61640.55 61859.19 505.87893
2021-10-31 02:00:00+02:00 61859.19 62328.38 61859.18 62328.13 1046.30598
2021-10-31 02:15:00+02:00 62328.13 62399.00 62142.74 62246.30 679.96187
2021-10-31 02:30:00+02:00 62246.30 62371.72 62171.01 62301.39 355.25273
2021-10-31 02:45:00+02:00 62301.39 62400.15 62220.27 62375.29 568.02797
2021-10-31 02:00:00+01:00 62375.28 62405.30 62260.01 62367.79 421.52566
2021-10-31 02:15:00+01:00 62367.79 62394.06 62219.50 62270.00 537.53843
2021-10-31 02:30:00+01:00 62270.00 62371.49 62156.83 62169.01 452.62728
2021-10-31 02:45:00+01:00 62169.01 62227.00 62114.18 62169.00 283.45948
:param int limit: The limit is the quantity of candles requested. Can combine with start_time or end_time:
- If no start or end timestamp passed, gets current timestamp and limit of candles backwards.
- If just start time or endtime passed, then applies the limit to find the endtime
or start time respectively.
- if endtime and start time passed, it gets all the ticks between, even if more than 1000 by calling
several times to API until completed. Limit ignored.
:param str time_zone: The index of the pandas dataframe in the object can be converted to any timezone, i.e. "Europe/Madrid"
- TZ database: https://en.wikipedia.org/wiki/List_of_tz_database_time_zones
The binance API use milliseconds from epoch (1970-01-01 00:00:00 UTC), that means that converting to a timezoned
date with daily time saving changes, the result would show the change in the timestamp.
Candles are ordered with the timestamp regardless of the index name, even if the index shows the hourly change
because daily time saving changes.
:param bool closed: The last candle is a closed one in the moment of the creation, discarding the current running one not closed
yet. Default True.
:param int display_columns: Number of columns in the dataframe display. Convenient to adjust in jupyter notebooks.
:param int display_max_rows: Number of rows in the dataframe display. Convenient to adjust in jupyter notebooks.
:param int display_width: Display width in the dataframe display. Convenient to adjust in jupyter notebooks.
:param int display_min_rows: Number of rows in the dataframe display. Convenient to adjust in jupyter notebooks.
:param bool or str postgres_klines: If True, gets data from a postgres database by selecting interactively from tables found.
Also a string with table name can be used. If str passed, it will be used as host.
:param bool postgres_agg_trades: If True, gets data from a postgres database by selecting interactively from tables found.
Also a string with table name can be used. If str passed, it will be used as host.
:param bool postgres_atomic_trades: If True, gets data from a postgres database by selecting interactively from tables found.
Also a string with table name can be used. If str passed, it will be used as host.
:param bool or str binbase: If True, gets data from binbase TimescaleDB (shared schema with symbol column).
If str passed, it will be used as host. Password from ``BINBASE_PASSWORD`` env var or panzer's ``binbase_password`` (~/.panzer_creds).
:param float hours: If hours is passed, it gets the candles from the last hours.
:param bool from_csv: If True, gets data from a csv file by selecting interactively from csv files found.
Also a string with filename can be used.
:param dict info_dic: Sometimes, for iterative processes, info_dic can be passed to avoid calling API for it. Its
weight is heavy.
:param bool or str from_csv: If True, gets data from a csv file by selecting interactively from csv files found.
Also a string with filename can be used.
:param dict info_dic: Sometimes, for iterative processes, info_dic can be passed to avoid calling API for it. Its
weight is heavy.
Examples:
.. code-block::
binpan_object = binpan.Symbol(symbol='ethbusd',
tick_interval='5m',
limit=100,
display_rows=10)
binpan_object.df
Open High Low Close Volume Quote volume Trades Taker buy base volume Taker buy quote volume
ETHBUSD 5m UTC
2022-06-22 08:55:00+00:00 1077.29 1081.11 1076.34 1080.18 810.1562 8.737861e+05 1401.0 410.0252 4.422960e+05
2022-06-22 09:00:00+00:00 1080.28 1087.31 1079.90 1085.03 2446.4199 2.653134e+06 3391.0 1253.6380 1.359514e+06
2022-06-22 09:05:00+00:00 1085.14 1085.98 1080.70 1080.99 1502.4415 1.628110e+06 1664.0 509.9937 5.525626e+05
2022-06-22 09:10:00+00:00 1080.79 1084.53 1080.26 1083.30 906.5092 9.814984e+05 1438.0 514.6718 5.573028e+05
2022-06-22 09:15:00+00:00 1083.42 1085.35 1081.41 1081.53 1191.9566 1.291633e+06 1549.0 736.5743 7.983178e+05
... ... ... ... ... ... ... ... ... ...
2022-06-22 16:50:00+00:00 1067.61 1070.00 1066.21 1069.32 1951.1378 2.084288e+06 1535.0 1396.4956 1.491744e+06
2022-06-22 16:55:00+00:00 1069.47 1078.61 1069.47 1077.78 2363.1639 2.539150e+06 3578.0 1239.7735 1.332085e+06
2022-06-22 17:00:00+00:00 1077.96 1079.32 1071.25 1073.00 1452.3712 1.560748e+06 2534.0 651.9175 7.006751e+05
2022-06-22 17:05:00+00:00 1072.76 1075.04 1071.00 1073.47 728.9761 7.820787e+05 1383.0 303.3896 3.254396e+05
2022-06-22 17:10:00+00:00 1073.42 1076.30 1073.24 1075.77 228.4269 2.456515e+05 426.0 127.7424 1.373634e+05
100 rows × 9 columns
"""
def __init__(self,
symbol: str = None,
tick_interval: str = None,
start_time: int | str = None,
end_time: int | str = None,
limit: int = 1000,
time_zone: str = 'Europe/Madrid',
closed: bool = True,
hours: float = None,
postgres_klines: bool | str = False,
postgres_agg_trades: bool | str = False,
postgres_atomic_trades: bool | str = False,
binbase: bool | str = False,
display_columns: int = 25,
display_max_rows: int = 10,
display_min_rows: int = 25,
display_width: int = 320,
from_csv: bool | str = False,
info_dic: dict = None):
"""
Create a Symbol instance for a Binance trading pair.
Fetches candlestick (kline) data from the Binance API and stores it as a pandas DataFrame.
The resulting object provides methods for technical indicators, plotting, trade analysis,
and strategy backtesting.
:param str symbol: Trading pair symbol (e.g. ``'BTCUSDT'``, ``'ethbtc'``). Case-insensitive.
:param str tick_interval: Candle timeframe (``'1m'``, ``'5m'``, ``'15m'``, ``'1h'``, ``'4h'``, ``'1d'``, etc.).
:param str | int start_time: Start of the time range. String (``'2024-01-01 00:00:00'``) or millisecond timestamp.
:param str | int end_time: End of the time range. String or millisecond timestamp.
:param int limit: Maximum number of candles to fetch. Default 1000.
:param str time_zone: IANA timezone for the DataFrame index. Default ``'Europe/Madrid'``.
:param bool closed: If True, only return closed candles. Default True.
:param float hours: Fetch the last N hours of data. Overrides ``limit`` when set.
:param bool | str postgres_klines: Load klines from PostgreSQL instead of API. Pass connection string or True.
:param bool | str postgres_agg_trades: Load aggregated trades from PostgreSQL.
:param bool | str postgres_atomic_trades: Load atomic trades from PostgreSQL.
:param bool | str binbase: Load data from binbase TimescaleDB. Pass True or host string.
:param int display_columns: Max columns to display in repr. Default 25.
:param int display_max_rows: Max rows to display. Default 10.
:param int display_min_rows: Min rows to display. Default 25.
:param int display_width: Display width in characters. Default 320.
:param bool | str from_csv: Load data from a CSV file. Pass filename or True for interactive selection.
:param dict info_dic: Pre-loaded exchange info dictionary (avoids API call).
Example:
.. code-block:: python
import binpan
sym = binpan.Symbol(symbol='BTCUSDT',
tick_interval='15m',
time_zone='Europe/Madrid',
limit=200)
sym.df # pandas DataFrame with OHLCV data
"""
if not symbol and not from_csv:
raise BinPanException(f"BinPan Exception: symbol needed or CSV file to load data using from_csv='filename.csv'")
if not from_csv and not symbol.isalnum():
binpan_logger.error(f"BinPan Exception: Ilegal characters in symbol: {symbol}")
# check correct tick interval passed
# if not from_csv:
# tick_interval = check_tick_interval(tick_interval)
# inicialización temprana para from_csv (se sobreescriben luego)
self.tick_interval = tick_interval
self.time_zone = time_zone
# dataframe columns
self.original_columns = binance_api_candles_cols
self.agg_trades_columns = agg_trades_api_map_columns
self.atomic_trades_columns = atomic_trades_api_map_columns
self.reversal_columns = reversal_columns_order
# time cols
self.time_cols = time_cols
self.dts_time_cols = dts_time_cols
# version
self.version = __version__
# set de rutas
self.cwd = os.getcwd()
# parámetros principales
self.symbol = symbol.upper() if symbol else ""
# estado de conexiones por defecto (se sobrescribe abajo para símbolos no-CSV);
# garantiza que _get_trades funcione también en símbolos creados desde CSV.
self.binbase = False
self.connection_binbase = self.cursor_binbase = None
self.postgres_agg_trades = self.postgres_atomic_trades = False
self.connection_klines = self.cursor_klines = None
self.connection_agg_trades = self.cursor_agg_trades = None
self.connection_atomic_trades = self.cursor_atomic_trades = None
if from_csv:
self._init_from_csv(from_csv=from_csv)
else:
##############
# timestamps #
##############
tick_interval = check_tick_interval(tick_interval) if tick_interval else None
self.closed = closed
# Si hours o start+end definen el rango, ignorar el limit por defecto (1000)
if hours is not None or (start_time is not None and end_time is not None):
limit = None
self.timeframe = Timeframe(start=start_time,
end=end_time,
timezone_IANA=time_zone,
tick_interval=tick_interval,
limit=limit,
hours=hours,
closed=closed)
self.start_time = self.timeframe.start.open
self.end_time = self.timeframe.end.close
self.tick_interval = self.timeframe.tick_interval
self.limit = self.timeframe.get_limit()
self.hours = self.timeframe.get_hours()
self.time_zone = str(self.timeframe.timezone_IANA)
# database
self._init_postgres_connections(postgres_klines=postgres_klines,
postgres_agg_trades=postgres_agg_trades,
postgres_atomic_trades=postgres_atomic_trades)
self._init_binbase_connection(binbase=binbase)
self.postgres_klines = postgres_klines
self.postgres_agg_trades = postgres_agg_trades
self.postgres_atomic_trades = postgres_atomic_trades
self.binbase = binbase
# pandas visualization settings
self._init_display_settings(display_columns=display_columns,
display_max_rows=display_max_rows,
display_min_rows=display_min_rows,
display_width=display_width)
# indicators and relevant data initialization #
self.support_lines = None # support levels from trades
self.resistance_lines = None # support levels from trades
self.sr_data_source = None # "atomic trades", "aggregated trades" or "klines"
self.raw_agg_trades = []
self.agg_trades = Trades(trade_type='agg', columns=self.agg_trades_columns, symbol=self.symbol,
time_zone=self.time_zone)
self.raw_atomic_trades = []
self.atomic_trades = Trades(trade_type='atomic', columns=self.atomic_trades_columns, symbol=self.symbol,
time_zone=self.time_zone)
self.min_height = 7
self.min_reversal = 4
self.reversal_agg_klines = pd.DataFrame(columns=self.reversal_columns)
self.reversal_atomic_klines = pd.DataFrame(columns=self.reversal_columns)
self.orderbook_value = None
self.orderbook = pd.DataFrame(columns=['Price', 'Quantity', 'Side'])
################
# plot control #
################
self.row_control = dict()
self.color_control = dict()
self.color_fill_control = dict()
self.plotting_volume_ma = 21
self.indicators_filled_mode = dict()
self.axis_groups = dict()
self.global_axis_group = 99
self.strategies = 0
self.row_counter = 1
self.strategy_groups = dict()
self.plot_splitted_serie_couples = {}
binpan_logger.debug(f"New instance of BinPan Symbol {self.version}: {self.symbol},"
f" {self.tick_interval}, limit={self.limit}, start={self.start_time},"
f" end={self.end_time}, {self.time_zone}, closed_candles={self.closed}")
#################
# query candles #
#################
if not from_csv:
self._init_from_api()
# update timestamps from data
self.timestamps = self.get_timestamps()
self.dates = self.get_dates()
self.start_time, self.end_time = self.timestamps
self.len = len(self.df)
##################
# exchange setup #
##################
if not info_dic: # for loop operations can be passed to avoid api weight overcome
try:
self.info_dic = get_info_dic()
except Exception as e:
binpan_logger.warning(f"Error trying get_info_dic() from API: {e}")
self.info_dic = dict()
else:
if not isinstance(info_dic, dict) or len(info_dic) == 0:
raise BinPanException("info_dic must be a non-empty dictionary")
self.info_dic = info_dic
try:
self.tickSize = self.info_dic[self.symbol]['filters'][0]['tickSize']
self.decimals = get_decimal_positions(self.tickSize)
self.pip = self.tickSize
self.order_filters = self.get_order_filters()
self.order_types = self.get_order_types()
self.permissions = self.get_permissions()
self.precision = self.get_precision()
except KeyError: # maybe no internet connectio but database available in local
self.tickSize = None
self.decimals = None
self.pip = None
self.order_filters = None
self.order_types = None
self.permissions = None
self.precision = None
# self.tickSize = self.info_dic[self.symbol]['filters'][0]['tickSize']
# self.decimals = get_decimal_positions(self.tickSize)
# self.pip = self.tickSize
#
# self.order_filters = self.get_order_filters()
# self.order_types = self.get_order_types()
# self.permissions = self.get_permissions()
# self.precision = self.get_precision()
self.market_profile_df = pd.DataFrame()
self.rolling_support_resistance_df = pd.DataFrame()
# init vertical lines over candles
self.red_timestamps = []
self.blue_timestamps = []
# check api continuity data and notify to user
self.discontinuities = check_continuity(df=self.df, time_zone=self.time_zone)
# ------------------------------------------------------------------ #
# __init__ helpers #
# ------------------------------------------------------------------ #
def _init_from_csv(self, from_csv: bool | str) -> None:
"""Load kline data from a CSV file and set instance attributes."""
(self.df,
self.symbol,
self.tick_interval,
self.time_zone,
self.start_time,
self.end_time,
self.closed,
self.limit) = csv_klines_setup(from_csv=from_csv,
symbol=self.symbol,
tick_interval=self.tick_interval,
cwd=self.cwd,
time_zone=self.time_zone)
def _init_from_api(self) -> None:
"""Fetch candle data from API, PostgreSQL or binbase and build the DataFrame."""
if self.binbase:
from .storage.binbase import get_klines as bb_get_klines
self.df = bb_get_klines(cursor=self.cursor_binbase,
symbol=self.symbol,
tick_interval=self.tick_interval,
start_time=self.start_time,
end_time=self.end_time,
time_zone=self.time_zone)
elif self.postgres_klines:
import binpan.postgresql as postgresql # solo para pycharm
self.table = postgresql.sanitize_table_name(f"{self.symbol.lower()}@kline_{self.tick_interval}")
self.df = postgresql.get_data_and_parse(cursor=self.cursor_klines,
table=self.table,
symbol=self.symbol,
tick_interval=self.tick_interval,
time_zone=self.time_zone,
start_time=self.start_time,
end_time=self.end_time,
data_type='kline')
else:
self.raw = get_candles_by_time_stamps(symbol=self.symbol,
tick_interval=self.tick_interval,
start_time=self.start_time,
end_time=self.end_time,
limit=self.limit)
self.df = parse_candles_to_dataframe(raw_response=self.raw,
columns=self.original_columns,
time_cols=self.time_cols,
symbol=self.symbol,
tick_interval=self.tick_interval,
time_zone=self.time_zone)
def _init_postgres_connections(self,
postgres_klines: bool | str,
postgres_agg_trades: bool | str,
postgres_atomic_trades: bool | str) -> None:
"""Set up PostgreSQL connections for klines, agg trades and atomic trades."""
if postgres_klines or postgres_agg_trades or postgres_atomic_trades:
import binpan.postgresql as postgresql
postgresql_port = get_secret("postgresql_port")
postgresql_user = get_secret("postgresql_user")
postgresql_database = get_secret("postgresql_database")
if postgres_klines:
if type(postgres_klines) == str:
binpan_logger.info(f"Postgres connection requested: {postgres_klines}")
postgresql_host_klines = postgres_klines
else:
postgresql_host_klines = get_secret("postgresql_host_klines")
self.connection_klines, self.cursor_klines = postgresql.setup(symbol=self.symbol,
tick_interval=self.tick_interval,
postgresql_host=postgresql_host_klines,
postgresql_user=postgresql_user,
postgresql_database=postgresql_database,
postgres_klines=True,
postgres_agg_trades=False,
postgres_atomic_trades=False,
postgresql_port=postgresql_port)
if postgres_agg_trades:
if type(postgres_agg_trades) == str:
postgresql_host_aggTrades = postgres_agg_trades
else:
postgresql_host_aggTrades = get_secret("postgresql_host_aggTrades")
self.connection_agg_trades, self.cursor_agg_trades = postgresql.setup(symbol=self.symbol,
tick_interval=self.tick_interval,
postgresql_host=postgresql_host_aggTrades,
postgresql_user=postgresql_user,
postgresql_database=postgresql_database,
postgres_klines=False,
postgres_agg_trades=True,
postgres_atomic_trades=False,
postgresql_port=postgresql_port)
if postgres_atomic_trades:
if type(postgres_atomic_trades) == str:
postgresql_host_trades = postgres_atomic_trades
else:
postgresql_host_trades = get_secret("postgresql_host_trades")
self.connection_atomic_trades, self.cursor_atomic_trades = postgresql.setup(symbol=self.symbol,
tick_interval=self.tick_interval,
postgresql_host=postgresql_host_trades,
postgresql_user=postgresql_user,
postgresql_database=postgresql_database,
postgres_klines=False,
postgres_agg_trades=False,
postgres_atomic_trades=True,
postgresql_port=postgresql_port)
else:
self.connection_klines, self.cursor_klines = None, None
self.connection_agg_trades, self.cursor_agg_trades = None, None
self.connection_atomic_trades, self.cursor_atomic_trades = None, None
def _init_binbase_connection(self, binbase: bool | str) -> None:
"""Set up connection to binbase TimescaleDB."""
if binbase:
from .storage.binbase import connect as bb_connect
host = binbase if isinstance(binbase, str) else None
kwargs = {"host": host} if host else {}
self.connection_binbase, self.cursor_binbase = bb_connect(**kwargs)
binpan_logger.info(f"Conectado a binbase: {kwargs.get('host', 'default')}")
else:
self.connection_binbase, self.cursor_binbase = None, None
def _init_display_settings(self,
display_columns: int,
display_max_rows: int,
display_min_rows: int,
display_width: int) -> None:
"""Configure pandas display options."""
self.display_max_columns = display_columns
self.display_max_rows = display_max_rows
self.display_min_rows = display_min_rows
self.display_width = display_width
self.set_display_max_columns(self.display_max_columns)
self.set_display_width(self.display_width)
self.set_display_min_rows(self.display_min_rows)
self.set_display_max_rows(self.display_max_rows)
def __repr__(self):
return str(self.df)
##################
# Show variables #
##################
def _save_dataframe_csv(self, df: pd.DataFrame, data_type: str,
empty_message: str | None, timestamped_filename: bool) -> None:
"""Internal helper that builds the filename and delegates to :func:`save_dataframe_to_csv`."""
if empty_message and df.empty:
binpan_logger.info(empty_message)
return
if timestamped_filename:
start, end = self.get_timestamps()
filename = f"{df.index.name.replace('/', '-')} {data_type} {start} {end}.csv"
else:
filename = f"{df.index.name.replace('/', '-')} {data_type}.csv"
save_dataframe_to_csv(filename=filename, data=df, timestamp=not timestamped_filename)
binpan_logger.info(f"Saved file {filename}")
[docs]
def save_csv(self, timestamped_filename: bool = True) -> None:
"""
Saves current klines DataFrame to a csv file.
:param bool timestamped_filename: Adds start and end timestamps to the name.
:return: None
"""
self._save_dataframe_csv(self.df, "klines", None, timestamped_filename)
[docs]
def save_atomic_trades_csv(self, timestamped_filename: bool = True) -> None:
"""
Saves current atomic trades to a csv file.
:param bool timestamped_filename: Adds start and end timestamps to the name.
:return: None
"""
self._save_dataframe_csv(self.atomic_trades.df, "atomicTrades", "No atomic trades to save.", timestamped_filename)
[docs]
def save_agg_trades_csv(self, timestamped_filename: bool = True) -> None:
"""
Saves current aggregated trades to a csv file.
:param bool timestamped_filename: Adds start and end timestamps to the name.
:return: None
"""
self._save_dataframe_csv(self.agg_trades.df, "aggTrades", "No aggregated trades to save.", timestamped_filename)
##################
# pandas display #
##################
[docs]
def set_display_max_columns(self, display_columns=None) -> None:
"""
Method to change the maximum number of columns shown in the display of the dataframe. Uses pandas options.
:param int display_columns: Integer
:return: None
"""
if display_columns:
self.display_max_columns = display_columns
pd.set_option('display.max_columns', display_columns)
else:
pd.set_option('display.max_columns', self.display_max_columns)
[docs]
def set_display_min_rows(self, display_min_rows=None) -> None:
"""
Method to change the number of minimum rows shown in the display of the dataframe. Uses pandas options.
:param int display_min_rows: Integer
"""
if display_min_rows:
self.display_min_rows = display_min_rows
pd.set_option('display.min_rows', display_min_rows)
else:
pd.set_option('display.min_rows', self.display_min_rows)
[docs]
def set_display_max_rows(self, display_max_rows=None) -> None:
"""
Method to change the number of maximum rows shown in the display of the dataframe. Uses pandas options.
:param int display_max_rows: Integer
"""
if display_max_rows:
self.display_max_rows = display_max_rows
pd.set_option('display.max_rows', display_max_rows)
else:
pd.set_option('display.max_rows', self.display_max_rows)
[docs]
def set_display_width(self, display_width: int = None) -> None:
"""
Method to change the width shown in the display of the dataframe. Uses pandas options.
:param int display_width: Integer
"""
if display_width:
self.display_width = display_width
pd.set_option('display.width', display_width)
else:
pd.set_option('display.width', self.display_width)
[docs]
@staticmethod
def set_display_decimals(display_decimals: int) -> None:
"""
Method to change the number of decimals shown in the display of the dataframe. Uses pandas options.
:param int display_decimals: Integer
"""
arg = f'%.{display_decimals}f'
pd.set_option('display.float_format', lambda x: arg % x)
###########
# methods #
###########
[docs]
def basic(self, exceptions: list = None, actions_col='actions') -> pd.DataFrame:
"""
Shows just a basic selection of columns data in the dataframe.
:param list exceptions: Columns names to keep.
:param str actions_col: To keep tags for buy or sell actions.
:return pd.DataFrame: Pandas DataFrame
"""
return basic_dataframe(data=self.df, exceptions=exceptions, actions_col=actions_col)
[docs]
def drop(self, columns_to_drop=None, inplace=False) -> pd.DataFrame:
"""
It drops some columns from the dataframe. If columns list not passed, then defaults to the initial columns.
Can be used when messing with indicators to clean the object.
:param: list columns: A list with the columns names to drop. If not passed, it defaults to the initial columns that remain
from when instanced. Defaults to any column but initial ones.
:param: bool inplace: When true, it drops columns in the object. False just returns a copy without that columns and dataframe
in the object remains.
:return pd.DataFrame: Pandas DataFrame with columns dropped.
"""
# if not columns_to_drop:
# columns_to_drop = []
current_columns = self.df.columns
if not columns_to_drop:
columns_to_drop = []
for col in current_columns:
if not col in self.original_columns:
columns_to_drop.append(col) # self.row_counter = 1
try:
if inplace:
conserve_columns = [c for c in current_columns if c not in columns_to_drop and c in self.row_control.keys()]
# conserve_columns = [c for c in current_columns if c not in columns_to_drop]
# clean strategy groups
clean_strategy_groups = {}
for k, v in self.strategy_groups.items():
clean_strategy_groups[k] = []
for col in v:
if col in conserve_columns:
clean_strategy_groups[k].append(col)
clean_strategy_groups = {k: v for k, v in clean_strategy_groups.items() if v}
self.strategy_groups = clean_strategy_groups
binpan_logger.debug(f"row_control: {self.row_control}")
self.row_control = {c: self.row_control[c] for c in conserve_columns}
extra_rows = set(self.row_control.values())
extra_rows.discard(1)
self.row_counter = 1 + len(extra_rows)
# reacondicionar rows de lo que ha quedado
unique_rows_index = {v: i + 2 for i, v in enumerate(extra_rows)}
binpan_logger.debug(f"conserve_columns: {conserve_columns} unique_rows_index: {unique_rows_index} extra_rows: {extra_rows}")
new_row_control = {}
for c in conserve_columns:
if self.row_control[c] in unique_rows_index.keys():
new_row_control.update({c: unique_rows_index[self.row_control[c]]})
else:
new_row_control.update({c: 1}) # not existing row for a line goes to 1 row
self.row_control = new_row_control
# self.row_control = {c: unique_rows_index[self.row_control[c]] if c in self.row_control.keys() else 1 for c in
# conserve_columns}
# clean plotting areas info
self.plot_splitted_serie_couples = {c: self.plot_splitted_serie_couples[c] for c in conserve_columns if
c in self.plot_splitted_serie_couples.keys()}
self.color_control = {c: self.color_control[c] for c in conserve_columns}
self.color_fill_control = {c: self.color_fill_control[c] for c in conserve_columns}
# revisar esto cuando el fill to next y esté hecho
self.indicators_filled_mode = {c: self.indicators_filled_mode[c] for c in conserve_columns if
c in self.indicators_filled_mode.keys()}
self.axis_groups = {c: self.axis_groups[c] for c in conserve_columns if c in self.axis_groups.keys()}
self.df = self.df.drop(columns_to_drop, axis=1)
return self.df
else:
return self.df.drop(columns_to_drop, axis=1, inplace=False)
except KeyError as exc:
wrong = (set(self.df.columns) | set(columns_to_drop)) - set(self.df.columns)
msg = f"BinPan Exception: Wrong column names to drop: {wrong} {exc}"
binpan_logger.error(msg)
raise Exception(msg)
[docs]
def delete_indicator_family(self, indicator_name_root: str) -> pd.DataFrame | None:
"""
Deletes indicator from dataframe. It search for plot info and also deletes other indicators in the same plot row level.
:param str indicator_name_root: Starting characters for the columns to delete. Case-insensitive.
:return: Resulting main dataframe.
"""
columns = [c for c in self.df.columns if c.startswith(indicator_name_root)]
if not columns:
binpan_logger.debug(f"BinPan Warning: No columns found with root '{indicator_name_root}': {self.df.columns}")
return
level = self.row_control[columns[0]]
columns = self.remove_plot_info_associated_columns(columns=columns, row_level=level)
for c in columns:
self.df = self.df.drop(c, axis=1)
return self.df
def _convert_indicator_data(
self,
source_data: pd.Series | pd.DataFrame | np.ndarray | list,
current_df: pd.DataFrame,
name: str | None,
names: list[str] | None,
suffix: str,
data_qty: int,
plotting_rows: list[int],
colors: list[str],
color_fills: list[str | bool],
) -> tuple[pd.Series | pd.DataFrame, list[pd.Series]] | None:
"""Convert heterogeneous input data into a (data, data_series) pair.
Handles pd.Series, pd.DataFrame, np.ndarray and list of them.
For list inputs the method recursively calls *insert_indicator* for each
element and returns ``None`` to signal an early exit. An unsupported
type also returns ``None`` after logging a warning.
"""
if type(source_data) == pd.Series:
data = source_data.copy(deep=True)
data.index = current_df.index
if name:
col_name = name
elif names:
col_name = names[0]
elif not data.name:
if suffix:
col_name = suffix
else:
col_name = f'Inserted_{len(self.df.columns)}' + suffix
else:
col_name = str(data.name) + suffix
data.name = col_name
return data, [data]
elif type(source_data) == pd.DataFrame:
data = source_data.copy(deep=True)
data = data.set_index(current_df.index)
data_series = [data[col] for col in data.columns]
for name_idx, ser in enumerate(data_series):
if names:
ser.name = names[name_idx]
else:
ser.name = str(ser.name) + suffix
return data, data_series
elif type(source_data) == np.ndarray:
raw = source_data.copy()
if names:
data_ser = pd.Series(data=raw, index=current_df.index, name=names[0])
elif suffix:
data_ser = pd.Series(data=raw, index=current_df.index, name=suffix)
else:
data_ser = pd.Series(data=raw, index=current_df.index, name=f'Inserted_{len(self.df.columns)}')
return data_ser, [data_ser]
elif type(source_data) == list:
assert len(source_data) == len(plotting_rows)
if name and not names:
names = [f"{name}_{i}" for i in range(data_qty)]
for element_idx, new_element in enumerate(source_data):
assert type(new_element) in [pd.Series, np.ndarray]
# noinspection PyUnresolvedReferences
data = new_element.copy(deep=True)
if not names:
try:
# noinspection PyUnresolvedReferences
current_name = new_element.name
except Exception:
current_name = f"Indicator_{len(self.df) + element_idx}{suffix}"
else:
current_name = names[element_idx]
self.insert_indicator(source_data=data, plotting_rows=[plotting_rows[element_idx]], colors=[
colors[element_idx]], color_fills=[color_fills[element_idx]], names=[current_name], suffix=suffix)
return None # signals early return with self.df
else:
msg = f"BinPan Warning: Unexpected data type {type(source_data)}, expected pd.Series, np.ndarray, pd.DataFrame or list of them."
binpan_logger.warning(msg)
return None # signals early return with None
def _setup_indicator_plot_info(
self,
data: pd.Series | pd.DataFrame,
data_series: list[pd.Series],
current_df: pd.DataFrame,
plotting_rows: list[int],
colors: list[str],
color_fills: list[str | bool],
no_overlapped_plot_rows: bool,
) -> pd.DataFrame:
"""Assign plot rows, colors and color-fills for new indicator columns.
Updates *current_df* in place by adding the series columns, then sets
plot metadata via ``set_plot_color``, ``set_plot_color_fill`` and
``set_plot_row``. Returns the updated DataFrame.
"""
if self.is_new(source_data=data, suffix=''): # suffix is added before this to names
if no_overlapped_plot_rows:
# downcast rows to available except 1 (overlap)
rows_tags = {row: i + self.row_counter + 1 for i, row in enumerate(sorted(list(set(plotting_rows))))}
else:
rows_tags = {row: row for row in plotting_rows}
plotting_rows = [rows_tags[r] if r != 1 else 1 for r in plotting_rows]
for i, serie in enumerate(data_series):
column_name = str(serie.name) # suffix is added before this to names
current_df.loc[:, column_name] = serie
self.set_plot_color(indicator_column=column_name, color=colors[i])
self.set_plot_color_fill(indicator_column=column_name, color_fill=color_fills[i])
self.set_plot_row(indicator_column=column_name, row_position=plotting_rows[i])
self.row_counter = max(plotting_rows)
self.df = current_df
return self.df
[docs]
def insert_indicator(self,
source_data: pd.Series | pd.DataFrame | np.ndarray | list,
strategy_group: str | None = None,
plotting_row: int | None = None,
plotting_rows: list[int] | None = None,
color: str | None = None,
no_overlapped_plot_rows: bool = True,
colors: list[str] | None = None,
color_fills: list[str | bool] | None = None,
name: str | None = None,
names: list[str] | None = None,
suffix: str = '') -> pd.DataFrame | None:
"""
Adds one or more indicators to the DataFrame in place.
:param source_data: The source data for the indicator(s). Can be a Series, DataFrame, ndarray, or list thereof.
:param strategy_group: (Optional) Name of the strategy group to tag the inserted data.
:param plotting_row: (Optional) The specific row for plotting a single series. '1' overlaps with candles; other values create
new rows.
:param plotting_rows: (Optional) List of rows for plotting each series. '1' means overlap; other integers determine separate row
positions.
:param color: (Optional) Color for plotting a single series.
:param no_overlapped_plot_rows: If True, avoids overlapping plot rows for multiple series.
:param colors: (Optional) List of colors for each series indicator. Defaults to random colors if not provided.
:param color_fills: (Optional) List of color fills (as strings) or False to avoid filling. Example: 'rgba(26,150,65,0.5)'.
:param name: (Optional) Name for a single inserted object.
:param names: (Optional) List of names for each column when multiple indicators are inserted.
:param suffix: Suffix to add to the new column name(s). If the source data is nameless, the suffix becomes the entire name.
:return: The modified DataFrame with new indicators added, or None if the operation fails.
Note: This function dynamically assigns plotting rows and colors if they are not explicitly provided. It handles different types
of input data for indicators and integrates them into the existing DataFrame.
"""
if type(source_data) == list:
data_qty = len(source_data)
elif type(source_data) == pd.DataFrame:
data_qty = len(source_data.columns)
else:
data_qty = 1
if plotting_row and not plotting_rows:
plotting_rows = [plotting_row]
if not plotting_rows:
plotting_rows = [2 for _ in range(data_qty)]
if color and not colors:
colors = [color]
if not colors:
colors = [choice(_plotting().plotly_colors) for _ in range(data_qty)]
if not color_fills:
color_fills = [False for _ in range(data_qty)]
current_df = self.df.copy(deep=True)
result = self._convert_indicator_data(
source_data=source_data, current_df=current_df, name=name, names=names,
suffix=suffix, data_qty=data_qty, plotting_rows=plotting_rows,
colors=colors, color_fills=color_fills,
)
if result is None:
# list case already handled via recursive calls, or unsupported type
return self.df if type(source_data) == list else None
data, data_series = result
self._setup_indicator_plot_info(
data=data, data_series=data_series, current_df=current_df,
plotting_rows=plotting_rows, colors=colors, color_fills=color_fills,
no_overlapped_plot_rows=no_overlapped_plot_rows,
)
# tag strategy group for columns
if strategy_group:
for d in data_series:
self.set_strategy_groups(column=str(d.name), group=strategy_group)
return self.df
[docs]
def hk(self, inplace=False) -> pd.DataFrame:
"""
It computes Heikin Ashi candles. Any existing indicator column will not be recomputed. It is recommended to drop any indicator
before converting candles to Heikin Ashi.
:param bool inplace: Change object dataframe permanently whe True is selected. False shows a copy dataframe.
:return pd.DataFrame: Pandas DataFrame
"""
df_ = self.df.copy(deep=True)
cols = ['Open', 'High', 'Low', 'Close']
heikin_ashi_df = df_.iloc[:][cols]
heikin_ashi_df['Close'] = (df_['Open'] + df_['High'] + df_['Low'] + df_['Close']) / 4
for i in range(len(df_)):
if i == 0:
heikin_ashi_df.iat[0, 0] = df_['Open'].iloc[0]
else:
heikin_ashi_df.iat[i, 0] = (heikin_ashi_df.iat[i - 1, 0] + heikin_ashi_df.iat[i - 1, 3]) / 2
heikin_ashi_df['High'] = heikin_ashi_df.loc[:, ['Open', 'Close']].join(df_['High']).max(axis=1)
heikin_ashi_df['Low'] = heikin_ashi_df.loc[:, ['Open', 'Close']].join(df_['Low']).min(axis=1)
df_[cols] = heikin_ashi_df[cols]
df_.index.name += ' HK'
if inplace:
self.df = df_
return self.df
else:
return df_
[docs]
def get_timestamps(self) -> tuple[int, int]:
"""
Get the first Open timestamp and the last Open timestamp.
:return tuple[int, int]: Tuple of (start_ms, end_ms) open timestamps.
"""
if self.df.empty:
binpan_logger.warning(f"BinPan Warning: Empty dataframe, no timestamps available.")
return 0, 0
start = self.df.iloc[0]['Open timestamp']
# end = self.df.iloc[-1]['Close timestamp']
end = self.df.iloc[-1]['Open timestamp']
binpan_logger.debug(f"From first Open timestamp {start} to last Open timestamp {end}")
return int(start), int(end)
[docs]
def get_dates(self) -> tuple[str, str]:
"""
Get the first Open timestamp and the last Open timestamp, and converts to timezoned dates.
:return tuple[str, str]: Tuple of (start_date, end_date) as formatted strings.
"""
start, end = self.get_timestamps()
ret_start = convert_milliseconds_to_str(ms=start, timezoned=self.time_zone)
ret_end = convert_milliseconds_to_str(ms=end, timezoned=self.time_zone)
binpan_logger.debug(f"From first Open date {ret_start} to last Open date {ret_end}")
return ret_start, ret_end
def _get_trades(self,
trade_type: str,
hours: int = None,
minutes: int = None,
startTime: int | str = None,
endTime: int | str = None,
time_zone: str = None,
from_csv: str = None) -> Trades:
"""
Generic trade fetching logic for both aggregated and atomic trades.
:param str trade_type: Either 'agg' or 'atomic'.
:param int hours: Last N hours of trades.
:param int minutes: Last N minutes of trades.
:param int or str startTime: Start timestamp or date string.
:param int or str endTime: End timestamp or date string.
:param str time_zone: IANA time zone string.
:param str from_csv: Path to CSV file, or truthy to auto-select.
:return: Pandas DataFrame with trades.
"""
# config by trade type
if trade_type == 'agg':
label = 'aggregated'
csv_filter = 'aggTrades'
csv_data_type = 'aggTrades'
id_col = 'Aggregate tradeId'
valid_columns = agg_trades_columns_from_binance
file_error_msg = "File do not seems to be Aggregated Trades File!"
postgres_flag = self.postgres_agg_trades
cursor = self.cursor_agg_trades
pg_table_suffix = 'aggTrade'
pg_data_type = 'aggTrade'
api_fn = get_historical_agg_trades
parse_fn = parse_agg_trades_to_dataframe
columns_attr = self.agg_trades_columns
raw_attr_name = 'raw_agg_trades'
trades_attr_name = 'agg_trades'
else:
label = 'atomic'
csv_filter = 'atomicTrades'
csv_data_type = 'atomicTrades'
id_col = 'Trade Id'
valid_columns = atomic_trades_columns_from_binance
file_error_msg = "File do not seems to be Atomic Trades File!"
postgres_flag = self.postgres_atomic_trades
cursor = self.cursor_atomic_trades
pg_table_suffix = 'trade'
pg_data_type = 'trade'
api_fn = get_historical_atomic_trades
parse_fn = parse_atomic_trades_to_dataframe
columns_attr = self.atomic_trades_columns
raw_attr_name = 'raw_atomic_trades'
trades_attr_name = 'atomic_trades'
# time range
if time_zone:
self.time_zone = time_zone
if startTime:
startTime = convert_str_date_to_ms(date=startTime, time_zone=self.time_zone)
if endTime:
endTime = convert_str_date_to_ms(date=endTime, time_zone=self.time_zone)
now_ms = int(time() * 1000)
# resolve the END of the window first (explicit endTime > symbol end > now)
if endTime:
curr_endTime = endTime
elif self.end_time:
curr_endTime = min(self.end_time, now_ms)
else:
curr_endTime = now_ms
# hours/minutes count back from the END of the window (the symbol's most recent data),
# so the request always has startTime < endTime and the trades align with the candles.
# (Counting back from "now" broke when the symbol's end_time was in the past.)
if hours:
curr_startTime = curr_endTime - (1000 * 60 * 60 * hours)
elif minutes:
curr_startTime = curr_endTime - (1000 * 60 * minutes)
elif startTime:
curr_startTime = startTime
else:
curr_startTime = self.start_time
st = convert_milliseconds_to_str(curr_startTime, timezoned=self.time_zone)
en = convert_milliseconds_to_str(curr_endTime, timezoned=self.time_zone)
binpan_logger.info(f"Requesting {label} trades between {st} and {en}")
# data source: CSV
if from_csv:
if type(from_csv) == str:
filename = from_csv
else:
filename = select_file(path=self.cwd, extension='csv', name_filter=csv_filter)
_, _, _, _, _, _ = extract_filename_metadata(filename=filename,
expected_data_type=csv_data_type,
expected_symbol=self.symbol,
expected_timezone=self.time_zone)
df_ = read_csv_to_dataframe(filename=filename,
index_col="Timestamp",
secondary_index_col=id_col,
symbol=self.symbol,
index_time_zone=self.time_zone)
for col in df_.columns:
if col not in valid_columns:
raise BinPanException(file_error_msg)
origin = 'csv'
# data source: binbase
elif self.binbase:
from .storage import binbase as bb
origin = 'binbase'
if trade_type == 'agg':
df_ = bb.get_agg_trades(cursor=self.cursor_binbase, symbol=self.symbol,
start_time=curr_startTime, end_time=curr_endTime, time_zone=self.time_zone)
else:
df_ = bb.get_trades(cursor=self.cursor_binbase, symbol=self.symbol,
start_time=curr_startTime, end_time=curr_endTime, time_zone=self.time_zone)
# data source: PostgreSQL (legacy)
elif postgres_flag:
from .storage.postgresql import get_data_and_parse, sanitize_table_name
origin = 'postgresql'
table = sanitize_table_name(f"{self.symbol.lower()}_{pg_table_suffix}")
df_ = get_data_and_parse(cursor=cursor, table=table, symbol=self.symbol,
tick_interval=self.tick_interval, time_zone=self.time_zone,
start_time=curr_startTime, end_time=curr_endTime, data_type=pg_data_type)
# data source: Binance API
else:
origin = 'binance_api'
try:
raw_data = api_fn(symbol=self.symbol, startTime=curr_startTime, endTime=curr_endTime)
except Exception as e:
binpan_logger.error(f"Error fetching {raw_attr_name}: {e} "
f"(if it is an auth error, check API key in ~/.panzer_creds)")
raw_data = []
setattr(self, raw_attr_name, raw_data)
df_ = parse_fn(response=raw_data, columns=columns_attr, symbol=self.symbol,
time_zone=self.time_zone, drop_dupes=id_col)
df_ = convert_to_numeric(data=df_)
trades_obj = Trades(df_, trade_type=trade_type, origin=origin, columns=columns_attr,
symbol=self.symbol, time_zone=self.time_zone, raw=getattr(self, raw_attr_name))
setattr(self, trades_attr_name, trades_obj)
return trades_obj
[docs]
def get_agg_trades(self,
hours: int = None,
minutes: int = None,
startTime: int | str = None,
endTime: int | str = None,
time_zone: str = None,
from_csv: str = None) -> Trades:
"""
Calls the API and creates another dataframe included in the object with the aggregated trades from API for the period of the
created object.
.. note::
If the object covers a long time interval, this action can take a relative long time. The BinPan library take care of the
API weight and can take a sleep to wait until API weight returns to a low value. This avoids ban from the API.
:param int hours: If passed, it use just last passed hours for the plot.
:param int minutes: If passed, it use just last passed minutes for the plot.
:param int or str startTime: If passed, it use just from the timestamp or date in format
(%Y-%m-%d %H:%M:%S: **2022-05-11 06:45:42**)) for the plot.
:param int or str endTime: If passed, it use just until the timestamp or date in format
(%Y-%m-%d %H:%M:%S: **2022-05-11 06:45:42**)) for the plot.
:param str time_zone: A time zone for time index conversion. Example: "Europe/Madrid"
:param str from_csv: If set, loads from a file.
:return: Pandas DataFrame
"""
return self._get_trades('agg', hours=hours, minutes=minutes, startTime=startTime,
endTime=endTime, time_zone=time_zone, from_csv=from_csv)
[docs]
def get_atomic_trades(self,
hours: int = None,
minutes: int = None,
startTime: int | str = None,
endTime: int | str = None,
time_zone: str = None,
from_csv: str = None) -> Trades:
"""
Calls the API and creates another dataframe included in the object with the atomic trades from API for the period of the
created object.
.. note::
If the object covers a long time interval, this action can take a relative long time. The BinPan library take care of the
API weight and can take a sleep to wait until API weight returns to a low value.
:param int hours: If passed, it use just last passed hours for the plot.
:param int minutes: If passed, it use just last passed minutes for the plot.
:param int or str startTime: If passed, it use just from the timestamp or date in format
(%Y-%m-%d %H:%M:%S: **2022-05-11 06:45:42**)) for the plot.
:param int or str endTime: If passed, it use just until the timestamp or date in format
(%Y-%m-%d %H:%M:%S: **2022-05-11 06:45:42**)) for the plot.
:param str time_zone: A time zone for time index conversion. Example: "Europe/Madrid"
:param str from_csv: If set, loads from a file.
:return: Pandas DataFrame
"""
return self._get_trades('atomic', hours=hours, minutes=minutes, startTime=startTime,
endTime=endTime, time_zone=time_zone, from_csv=from_csv)
[docs]
def is_new(self, source_data: pd.Series | pd.DataFrame, suffix: str = '') -> bool:
"""
Verify if indicator columns are previously created to avoid allocating new rows and colors etc.
:param pd.Series or pd.DataFrame source_data: Data from pandas_ta to review if is previously computed.
:param str suffix: If suffix passed, it takes it into account when searching for existence.
:return bool:
"""
# existing_columns = list(self.df.columns)
if type(source_data) == np.ndarray:
return True
source_data = source_data.copy(deep=True)
generated_columns = []
if type(source_data) == pd.Series:
serie_name = str(source_data.name) + suffix
generated_columns.append(serie_name)
elif type(source_data) == pd.DataFrame:
generated_columns = [c + suffix for c in list(source_data.columns)]
else:
msg = f"BinPan error: (is_new?) source data is not pd.Series or pd.DataFrame"
binpan_logger.error(msg)
raise Exception(msg)
for gen_col in generated_columns:
if gen_col in self.df.columns:
binpan_logger.info(f"Existing column: {gen_col} -> No data added to instance.")
return False
else:
binpan_logger.info(f"New column: {gen_col}")
return True
[docs]
def get_reversal_agg_candles(self, min_height: int = 7, min_reversal: int = 4) -> pd.DataFrame | None:
"""
Resamples aggregated API trades to reversal klines:
https://atas.net/atas-possibilities/charts/how-to-set-reversal-charts-for-finding-the-market-reversal/
:param min_height: Defaults to 7. Minimum reversal kline height to close a candle
:param min_reversal: Defaults to 4. Minimum reversal from hig/low to close a candle
:return pd.DataFrame: Resample trades to reversal klines. Can be plotted.
"""
if self.agg_trades.empty:
binpan_logger.info(empty_agg_trades_msg)
return
if min_height:
self.min_height = min_height
if min_reversal:
self.min_reversal = min_reversal
if self.reversal_agg_klines.empty or min_height or min_reversal:
self.reversal_agg_klines = reversal_candles(trades=self.agg_trades.df, decimal_positions=self.decimals,
time_zone=self.time_zone, min_height=self.min_height,
min_reversal=self.min_reversal)
return self.reversal_agg_klines
[docs]
def get_reversal_atomic_candles(self, min_height: int = 7, min_reversal: int = 4) -> pd.DataFrame | None:
"""
Resamples API atomic trades to reversal klines:
https://atas.net/atas-possibilities/charts/how-to-set-reversal-charts-for-finding-the-market-reversal/
:param min_height: Defaults to 7. Minimum reversal kline height to close a candle
:param min_reversal: Defaults to 4. Minimum reversal from hig/low to close a candle
:return pd.DataFrame: Resample trades to reversal klines. Can be plotted.
"""
if self.atomic_trades.empty:
binpan_logger.info(empty_atomic_trades_msg)
return
if min_height:
self.min_height = min_height
if min_reversal:
self.min_reversal = min_reversal
if self.reversal_atomic_klines.empty or min_height or min_reversal:
self.reversal_atomic_klines = reversal_candles(trades=self.atomic_trades.df, decimal_positions=self.decimals,
time_zone=self.time_zone, min_height=self.min_height,
min_reversal=self.min_reversal)
return self.reversal_atomic_klines
[docs]
def resample(self, tick_interval: str, inplace=False) -> pd.DataFrame | None:
"""
Resample trades to a different tick interval. Tick interval must be higher.
:param str tick_interval: A binance tick interval. Must be higher than current tick interval.
:param bool inplace: Change object dataframe permanently whe True is selected. False shows a copy dataframe.
:return pd.DataFrame: Resampled klines.
"""
current_index = tick_interval_values.index(self.tick_interval)
new_index = tick_interval_values.index(tick_interval)
if new_index <= current_index:
binpan_logger.error(f"BinPan error: resample target '{tick_interval}' must be a HIGHER interval than "
f"the current '{self.tick_interval}' (klines can only be aggregated into coarser candles).")
return
binpan_logger.info(f"Resampling {self.symbol} from {self.tick_interval} to {tick_interval}")
if inplace:
self.drop(inplace=True)
self.tick_interval = tick_interval
self.row_control = dict()
self.color_control = dict()
self.color_fill_control = dict()
self.indicators_filled_mode = dict()
self.axis_groups = dict()
self.global_axis_group = 99
self.strategies = 0
self.row_counter = 1
self.strategy_groups = dict()
self.plot_splitted_serie_couples = {}
self.timestamps = self.get_timestamps()
self.dates = self.get_dates()
self.start_time, self.end_time = self.timestamps
from .analysis.aggregations import resample_klines
self.df = resample_klines(data=self.df, tick_interval=tick_interval)
self.discontinuities = check_continuity(df=self.df, time_zone=self.time_zone)
return self.df
else:
from .analysis.aggregations import resample_klines
return resample_klines(data=self.df, tick_interval=tick_interval)
[docs]
def repair_continuity(self) -> None:
"""
Repair kline discontinuities by filling missing candles.
Updates the instance DataFrame in place and re-checks continuity.
"""
self.df = repair_kline_discontinuity(df=self.df, time_zone=self.time_zone)
binpan_logger.info(f"Klines continuity verification after repair")
# verify discontinuity
self.discontinuities = check_continuity(df=self.df, time_zone=self.time_zone)
if self.discontinuities.empty:
binpan_logger.info(f"Klines continuity OK")
#################
# Exchange data #
#################
[docs]
def update_info_dic(self) -> dict:
"""
Returns exchangeInfo data when instantiated. It includes, filters, fees, and many other data for all symbols in the
exchange.
:return dict: Full exchange info dictionary for all symbols.
"""
self.info_dic = get_info_dic()
return self.info_dic
[docs]
def get_order_filters(self) -> dict:
"""
Get exchange info about the symbol for order filters.
:return dict: Dictionary of order filters for the symbol.
"""
filters = get_symbols_filters(info_dic=self.info_dic)
self.order_filters = filters[self.symbol]
return self.order_filters
[docs]
def get_order_types(self) -> list:
"""
Get exchange info about the symbol for order types.
:return list: List of available order types for the symbol.
"""
order_types_precision = get_orderTypes_and_permissions(info_dic=self.info_dic)
self.order_types = order_types_precision[self.symbol]['orderTypes']
return self.order_types
[docs]
def get_permissions(self) -> list:
"""
Get exchange info about the symbol for trading permissions.
:return list: Trading permissions for the symbol.
"""
permissions = get_orderTypes_and_permissions(info_dic=self.info_dic)
self.permissions = permissions[self.symbol]['permissions']
return self.permissions
[docs]
def get_precision(self) -> dict:
"""
Get exchange info about the symbol for assets precision.
:return dict: Precision settings for base and quote assets.
"""
precision = get_precision(info_dic=self.info_dic)
self.precision = precision[self.symbol]
return self.precision
[docs]
def get_api_status(self) -> str:
"""
Return the symbol status, TRADING, BREAK, etc.
:return str: Current trading status (e.g. ``'TRADING'``, ``'BREAK'``).
"""
return Exchange().df.loc[self.symbol].to_dict()['status']
######################
[docs]
def get_fees(self, symbol: str = None) -> dict:
"""
Shows applied fees for the symbol of the object.
Requires Binance API credentials, managed by panzer (~/.panzer_creds). panzer prompts for them on first use if missing.
:param symbol: Not to use it, just here for initializing the class.
:return: Dictionary
"""
if not symbol:
symbol = self.symbol
return get_fees(symbol=symbol, decimal_mode=False)
[docs]
def get_orderbook(self, limit: int = 5000) -> pd.DataFrame:
"""
Gets orderbook.
:param int limit: Defaults to maximum: 5000
:return pd.DataFrame:
"""
orders = get_order_book(symbol=self.symbol, limit=limit)
bids = pd.DataFrame(data=orders['bids'], columns=['Price', 'Quantity']).astype(float)
bids.loc[:, 'Side'] = 'bid'
asks = pd.DataFrame(data=orders['asks'], columns=['Price', 'Quantity']).astype(float)
asks.loc[:, 'Side'] = 'ask'
ob = pd.concat([bids, asks]).sort_values(by=['Price'], ascending=False).reset_index(drop=True)
ob.index.name = f"{self.symbol} updateId:{orders['lastUpdateId']}"
self.orderbook = ob
return self.orderbook