import pandas as pd
import numpy as np
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from crypto_env.types import Transaction
[docs]class Recorder:
"""
This is the recorder class. A recorder object is able to record agent's action in both training and production mod. The data collected can later be used for plotting or analyzing.
"""
[docs] def __init__(self, price_list, crypto_cap=0, fiat_cap=1000) -> None:
"""__init__
Args:
price_list (list): a list of price, the length should be equal to the size of the :py:class:`DataLoader`.
crypto_cap (int, optional): initial balance of crypto. Defaults to 0.
fiat_cap (int, optional): initial balance of fiat. Defaults to 1000.
"""
self._transaction_record = list()
self._info_record = list()
self._idx = -1
self._indexes = None
self._price_list = price_list
[docs] def reset(self):
"""Reset the recorder
"""
self._transaction_record = list()
self._info_record = list()
self._idx = 0
self._indexes = None
[docs] def insert_transaction(self, transaction: Transaction):
"""Insert new transaction into the recorder
Args:
transaction (:py:class:`.Transaction`): :py:class:`.Transaction` object to insert
"""
self._transaction_record.append(transaction)
self._idx += 1
[docs] def insert_info(self, info):
"""Insert market information into the record
Args:
info (array-like): an array of current market info
"""
if self._indexes is None:
self._indexes = info.index
self._info_record.append(list(info))
[docs] def get_transaction_record(self, idx=None):
"""Return all history buy and sell signals generated by the agent
Args:
idx (int, optional): Number of records to print. Defaults to None.
Returns:
DataFrame: A :py:class:`pandas.DataFrame` containing all history signals.
"""
if idx is None:
idx = self._idx
return pd.DataFrame(self._transaction_record).iloc[0:idx]
[docs] def get_info_record(self, to_dataframe=True):
"""Return all history market info
Args:
to_dataframe (bool, optional): Whether to convert to dataframe. Defaults to True.
Returns:
(DataFrame, any): Market information history
"""
if to_dataframe:
return pd.DataFrame(self._info_record, columns=self._indexes)
return self._info_record
[docs] def get_expenditure(self, idx=None):
"""Calculate how many fiat was used in the investment.
Args:
idx (int, optional): Number of transactions to involve. Defaults to None.
Returns:
float
"""
if idx is None:
idx = self._idx
transaction_record = self.get_transaction_record(idx)
buy_record = transaction_record[transaction_record['signal'] == 0]
buy_index = buy_record.index.to_list()
buy_amount = buy_record.value.to_numpy()
buy_price = self._price_list.iloc[buy_index].to_numpy()
total_expenditure = np.multiply(buy_amount, buy_price).sum()
return total_expenditure
[docs] def get_income(self, idx=None):
"""Calculate how many value does the agent earn.
Args:
idx (int, optional): How many transactions to involve. Defaults to None.
Returns:
float
"""
if idx is None:
idx = self._idx
transaction_record = self.get_transaction_record(idx)
sell_record = transaction_record[transaction_record['signal'] == 1]
sell_index = sell_record.index.to_list()
sell_amount = sell_record.value.to_numpy()
sell_price = self._price_list.iloc[sell_index].to_numpy()
total_income = np.multiply(sell_amount, sell_price).sum()
return total_income
[docs] def get_fiat_balance(self, idx=None):
"""Calculate fiat balance
Args:
idx (int, optional): How many transactions to involve. Defaults to None.
Returns:
float
"""
if idx is None:
idx = self._idx
return self.get_income(idx) - self.get_expenditure(idx)
[docs] def get_crypto_balance(self, idx=None):
"""Calculate cryptocurrency balance
Args:
idx (int, optional): How many transactions to involve. Defaults to None.
Returns:
_type_: _description_
"""
if idx is None:
idx = self._idx
transaction_record = self.get_transaction_record(idx)
sell_amount = transaction_record[transaction_record['signal'] == 1].value.sum()
buy_amount = transaction_record[transaction_record['signal'] == 0].value.sum()
return buy_amount - sell_amount
[docs] def get_crypto_value(self, idx=None):
"""Calculate the value of crypto in balance
Args:
idx (int, optional): How many transactions to involve. Defaults to None.
Returns:
float
"""
if idx is None:
idx = self._idx
crypto_balance = self.get_crypto_balance(idx)
crypto_value = crypto_balance * self._price_list.iloc[idx]
return crypto_value
[docs] def get_roi(self, idx=None):
"""Calculate the return of investment
Args:
idx (int, optional): How many transactions to involve. Defaults to None.
Returns:
float
"""
if idx is None:
idx = self._idx
net_return = self.get_crypto_value(idx) + self.get_fiat_balance(idx)
cost = self.get_expenditure(idx)
return net_return / (cost + 0.0001)