Reference

coin_test.data

Data loading / processing module.

class coin_test.data.BinanceDataset(name, asset_pair, freq='d', start=None, end=None)

Create datasets from downloaded Binance data.

property metadata: MetaData: Get metadata.

class coin_test.data.Composer(datasets, length)

Manages datasets for simulation.

get_lookback(timestamp, lookback, keys=None, mask=True)

Get lookback of data.

Wrapper for get_range. Convert integer number of lookback timesteps into a time range based of Composer's freq attribute.

Parameters:

timestamp (Timestamp) -- Starting timestamp of lookback.
lookback (int) -- Number of timesteps to lookback.
keys (Optional[Iterable[AssetPair]]) -- Optional. AssetPairs to filter results on.
mask (bool) -- Optional. Whether to mask non-Open data to NaN.

Returns:

Dictionary mapping asset pairs to retrieved rows of data per: asset pair. Dictionary default to all datasets, but is filtered based on keys parameter.

Return type:

dict

get_range(start_time, end_time, keys=None, mask=True)

Get range of data.

Parameters:

start_time (Timestamp) -- Starting timestamp of range.
end_time (Timestamp) -- Ending timestamp of range.
keys (Optional[Iterable[AssetPair]]) -- Optional. AssetPairs to filter results on.
mask (bool) -- Optional. Whether to mask non-Open data to NaN.

Returns:

Dictionary mapping asset pairs to retrieved rows of data per: asset pair. Dictionary default to all datasets, but is filtered based on keys parameter.

Return type:

dict

get_timestep(timestamp, keys=None, mask=True)

Get single timestep of data.

Parameters:

timestamp (Timestamp) -- Timestamp to get data for.
keys (Optional[Iterable[AssetPair]]) -- Optional. AssetPairs to filter results on.
mask (bool) -- Optional. Whether to mask non-Open data to NaN.

Returns:

Dictionary mapping asset pairs to timestamp data per asset: pair. Dictionary default to all datasets, but is filtered based on keys parameter.

Return type:

dict

class coin_test.data.CustomDataset(name, df, freq, pair, synthetic=False)

Load a DataFrame in the expected format of price data.

property metadata: MetaData: Price data metadata.

class coin_test.data.Datasaver(name, datasets)

Hold collections of datasets to save.

static load(fp)

Load Datasaver from disk.

Parameters:: fp (str) -- filepath to load from
Returns:: Datasaver stored at the location
Return type:: Datasaver
Raises:: ValueError -- raises ValueError if the specified file path is not a file

save(directory)

Pickle a Datasaver to disk.

Parameters:: directory (str) -- Filepath to save to
Returns:: _description_
Return type:: str

class coin_test.data.Dataset(*args, **kwargs)

Load some data into a DataFrame.

Use to provide the Dataset class a consistent interface to access data loaded from different sources. Note that extending classes MUST set the df attribute in their __init__ method, else an error will be raised. Also note that setting df will trigger validation of the new dataframe.

process(processors)

Process the dataset.

Return type:: Dataset

split(timestamp=None, length=None, percent=None, pre_name='_pre', post_name='_post')

Split Dataset into Pre and Post split Datasets.

Parameters:

timestamp (Optional[Timestamp]) -- [Optional] Timestamp to split Dataset on
length (Optional[Timedelta]) -- [Optional] pd.Timedelta to specify length of the pre-split dataset
percent (Optional[float]) -- [Optional] float percentage of the data to split on
pre_name (str) -- Suffix to append to dataset name for pre section of the split
post_name (str) -- Suffix to append to dataset name for post section of the split

Returns:

Pre and post split datasets of the same type as the original dataset

Return type:

tuple

class coin_test.data.DatasetGenerator

Create synthetic datasets.

abstract generate(timedelta, seed=None, n=1)

Create synthetic datasets from the given dataset.

Parameters:

timedelta (Timedelta) -- A time range for the new datasets
seed (Optional[int]) -- A random seed for the generated datasets
n (int) -- The number of datasets to generate

Returns:

The synthetic datasets

Return type:

list[DATASET_TYPE]

class coin_test.data.FillProcessor(freq, method='pad'): Fill missing periods for a dataset.

class coin_test.data.GarchDatasetGenerator(dataset, chunk_size=1, mean='Constant', lags=0, vol='GARCH', p=1, o=0, q=1, power=2, dist='normal', hold_back=None, rescale=None)

Synthetic Dataset Generator with GARCH.

Use Generalized Autoregressive Conditional Heteroskedasticity (GARCH) model. Default values initialize a GARCH(1,1) model with constant mean.

Close prices are simulated with univariate GARCH model. Open prices are set as previous day's close. High and Low prices are randomly sampled chunks transformed into and reverted from percent changes relative to min/max Open/Close of each period bar.

DATASET_TYPE: alias of CustomDataset

static from_pct_change_based_on_extreme_bar_open_and_close(pct_change_series, extreme, open_series, close_series)

Calculates values from percent change and min/max of Open and Close.

Parameters:

pct_change_series (Series) -- a series of percent change to turn into values
extreme (Literal['max', 'min']) -- selects either min or max when calculating extreme series
open_series (Series) -- series of open price values
close_series (Series) -- series of close price values

Returns:

extreme_series + extreme_series * pct_change_series

Return type:

series

Raises:

ValueError -- open_series should be same length as close_series.
ValueError -- pct_change_series should be same length as open_series.

static from_pct_change_based_on_starting_value(pct_change_series, starting_price)

Generate synthetic series from GARCH model fit to univariate data.

Return type:: Series

generate(timedelta, seed=None, n=1)

Create synthetic datasets from GARCH model fit to given dataset.

Parameters:

timedelta (Timedelta) -- A time range for the new datasets
seed (Optional[int]) -- A random seed for the generated datasets
n (int) -- The number of datasets to generate

Returns:

The synthetic datasets

Return type:

list[PriceDataset]

static get_garch_model_parameters(univariate_series, garch_settings)

Gets GARCH model parameters estimated from given univariate series.

Parameters:

univariate_series (Series) -- series of data
garch_settings (GarchSettings) -- arguments to contruct the GARCH model with

Returns:

dictionary of model parameters: estimated from fitting to univariate series.

Return type:

res_garch_model.params

static sample_series(series, num_rows, chunk_size, rng)

Randomly samples a new series from a given series.

Parameters:

series (Series) -- series to sample from
num_rows (int) -- The number of rows in the new series
chunk_size (int) -- The amount of rows to combine in each chunk
rng (Generator) -- A numpy random number generator

Returns:

a new sampled series

Return type:

pd.Series

static to_pct_change_based_on_extreme_bar_open_and_close(series_to_pct_change, extreme, open_series, close_series)

Calculates percent change relative to min/max of Open and Close.

Parameters:

series_to_pct_change (Series) -- a series of values to turn into percent change
extreme (Literal['max', 'min']) -- selects either min or max when calculating extreme series
open_series (Series) -- series of open price values
close_series (Series) -- series of close price values

Returns:

difference of series_to_pct_change and extreme_series as proportion of extreme_series.

Return type:

pct_change_series

class coin_test.data.GarchSettings(mean, lags, vol, p, o, q, power, dist, hold_back, rescale): Class for keeping track of settings to intialize a GARCH model.

class coin_test.data.IdentityProcessor: Identity Transform.

class coin_test.data.MetaData(pair: AssetPair, freq: str)

Historical data metadata.

freq: str: Alias for field number 1

pair: AssetPair: Alias for field number 0

class coin_test.data.PriceDataset(*args, **kwargs)

Load a DataFrame with associated MetaData.

abstract property metadata: MetaData: The dataframe metadata.

class coin_test.data.Processor: Transform a DataFrame.

class coin_test.data.ReturnsDatasetGenerator(dataset)

Create synthetic datasets by shuffling the percentage gains each day.

DATASET_TYPE: alias of CustomDataset

generate(timedelta, seed=None, n=1)

Create returns-based synthetic datasets from the given dataset.

Parameters:

timedelta (Timedelta) -- A time range for the new datasets
seed (Optional[int]) -- A random seed for the generated datasets
n (int) -- The number of datasets to generate

Returns:

The synthetic datasets

Return type:

list[DATASET_TYPE]

static select_data(df_norm, starting_price, num_rows, rng)

Take a normalized Dataframe and create a synthetic dataset from it.

Parameters:

df_norm (DataFrame) -- Normalized Dataframe of original data
starting_price (float) -- The first open price for the data
num_rows (int) -- The number of rows in the dataset
rng (Generator) -- A numpy random number generator

Returns:

The synthetic dataset

Return type:

pd.DataFrame

class coin_test.data.SamplingDatasetGenerator

ABC for sampling dataset generators.

static create_index(start, timedelta, freq)

Create a PeriodIndex given a start time, timedelta, and frequency.

Return type:: PeriodIndex

static normalize_row_data(df)

Normalize the row data so that it can be sampled with returns.

Return type:: DataFrame

static unnormalize(synthetic_df)

Take a normalized Dataframe and unnormalize it.

Essentially, convert columns from representing percentage increases to actual prices.

Parameters:: synthetic_df (DataFrame) -- Normalized Dataframe
Returns:: The unnormalized Dataframe
Return type:: pd.DataFrame

class coin_test.data.StitchedChunkDatasetGenerator(dataset, chunk_size=10)

Synthetic Dataset Generator with chunks of data.

DATASET_TYPE: alias of CustomDataset

generate(timedelta, seed=None, n=1)

Create chunk-based synthetic datasets from the given dataset.

Parameters:

timedelta (Timedelta) -- A time range for the new datasets
seed (Optional[int]) -- A random seed for the generated datasets
n (int) -- The number of datasets to generate

Returns:

The synthetic datasets

Return type:

list[DATASET_TYPE]

static select_data(df_norm, starting_price, num_rows, chunk_size, rng)

Take a normalized Dataframe and create a synthetic dataset from it.

Parameters:

df_norm (DataFrame) -- Normalized Dataframe of original data
starting_price (float) -- The first open price for the data
num_rows (int) -- The number of rows in the dataset
chunk_size (int) -- The amount of rows to combine in each chunk
rng (Generator) -- A numpy random number generator

Returns:

The synthetic dataset

Return type:

pd.DataFrame

class coin_test.data.WindowStepDatasetGenerator(dataset)

Windows of data as separate datasets.

DATASET_TYPE: alias of CustomDataset

static calc_window_length(freq, timedelta)

Calculate the number of rows in each window of length timedelta.

Return type:: int

static extract_windows(df_total, freq, timedelta, n)

Take a DataFrame and create windows from it.

Parameters:

df_total (DataFrame) -- Entire original DataFrame
freq (str) -- The frequency of the DataFrame PeriodIndex
timedelta (Timedelta) -- The length in time per window
n (int) -- The number of windows

Returns:

The windows

Return type:

list[pd.DataFrame]

generate(timedelta, seed=None, n=1)

Create uniformly distributed window+step datasets.

Given a length of time for each dataset, along with a number of datasets to create, make new datasets of a given length that are equally spaced from each other, creating overlapping datasets if necessary.

Parameters:

timedelta (Timedelta) -- A time range for the new datasets
seed (Optional[int]) -- Irrelevant for this implementation
n (int) -- The number of datasets to generate

Returns:

The synthetic datasets

Return type:

list[DATASET_TYPE]

static make_slices(total_length, window_length, n)

Given the total dataset length and window length, make slices for the df.

Return type:: list[slice]

coin_test.backtest

Backtesting module of the coin-test library.

class coin_test.backtest.BacktestResults(composer, starting_portfolio, strategies, sim_data, slippage_calculator_type, transaction_fee_calculator_type)

Record the results of a backtest.

static create_date_price_df(sim_data, composer)

Create a TimeSeries dataframe for portfolio value over time.

Return type:: Series

static load(fp)

Load BacktestResults from disk.

Parameters:: fp (str) -- filepath to pickle file to load from
Returns:: BacktestResults stored at the location
Return type:: BacktestResults
Raises:: ValueError -- raises ValueError if the specified file path is not a file

save(path)

Save to disk.

Parameters:: path (str) -- Path to save to.
Return type:: None

static value_from_portfolio(t, p, c)

Get the monetary value of a portfolio.

Return type:: float

class coin_test.backtest.ConstantSlippage(basis_points=50.0): A Constant slippage Calculator.

class coin_test.backtest.ConstantTransactionFeeCalculator(basis_points=50.0): Calculate Constant the transactions fees for a trade.

class coin_test.backtest.GaussianSlippage(rng, mean_bp, std_dev_bp): A Constant slippage Calculator.

class coin_test.backtest.LimitTradeRequest(asset_pair, side, limit_price, notional=None, qty=None)

A TradeRequest implementation for limit orders.

If buying, buy when the current price is less than the limit price. If selling, sell when the current price is greater than the limit price.

should_execute(price)

Execute when the limit price condition is reached.

Return type:: bool

class coin_test.backtest.MarketTradeRequest(asset_pair, side, notional=None, qty=None)

A TradeRequest implementation for market (GTC) orders.

build_trade(current_asset_price, slippage_calculator, transaction_fee_calculator)

Build Trade that represents a TradeRequest for a MarketTradeRequest.

Parameters:

current_asset_price (dict[AssetPair, DataFrame]) -- Current price data from composer
slippage_calculator (SlippageCalculator) -- Slippage Calculator implementation
transaction_fee_calculator (TransactionFeeCalculator) -- TransactionFeeCalculator implementation

Return type:

Trade

Returns:

Trade that the TradeRequest represents

should_execute(price)

A MarketTrade object should always execute.

Return type:: bool

class coin_test.backtest.Portfolio(base_currency, assets)

Manage a portfolio.

adjust(trade)

Adjust the portfolio after a given Trade is performed.

Parameters:: trade (Trade) -- The Trade object that is completed
Returns:: The new adjusted portfolio or None if insufficient Money
Return type:: Portfolio | None

available_assets(asset)

Return the available assets for a given ticker in a portfolio.

Parameters:: asset (Ticker) -- The desired ticker
Returns:: The amount of money available for the given ticker
Return type:: Money
Raises:: ValueError -- If the ticker is not in the portfolio

class coin_test.backtest.Simulator(composer, starting_portfolio, strategies, slippage_calculator, transaction_fee_calculator, warn_on_error=True)

Manage the simulation of a backtest.

run()

Run a simulation.

Return type:: BacktestResults

run_strategies(schedule, time, portfolio)

Create TradeRequests for a given timestamp.

Parameters:

schedule (Iterable[tuple[Strategy, croniter]]) -- List of strategies indicating their next run time
time (Timestamp) -- Current timestamp used to determine which strategies should run
portfolio (Portfolio) -- Current Portfolio at given timestamp

Raises:

ValueError -- If a strategy raises an error and warn_on_error is False

Return type:

list[TradeRequest]

Returns:

list of TradeRequests to handle

class coin_test.backtest.SlippageCalculator: Calculate the slippage of an asset.

class coin_test.backtest.StopLimitTradeRequest(asset_pair, side, stop_limit_price, notional=None, qty=None)

A TradeRequest implementation for stop limit orders.

If buying, buy when the current price is greater than the limit price. If selling, sell when the current price is less than the limit price.

should_execute(price)

Execute when the stop limit price condition is reached.

Return type:: bool

class coin_test.backtest.Strategy(name, asset_pairs, schedule, lookback): Strategy generates TradeRequests.

class coin_test.backtest.Trade(asset_pair, side, amount, price, transaction_fee=0): Store the details of a trade.

class coin_test.backtest.TradeRequest(asset_pair, side, notional=None, qty=None)

Request a trade with given specifications.

abstract build_trade(current_asset_price, slippage_calculator, transaction_fee_calculator)

Build Trade that represents a TradeRequest.

Parameters:

current_asset_price (dict[AssetPair, DataFrame]) -- Current price data from composer
slippage_calculator (SlippageCalculator) -- Slippage Calculator implementation
transaction_fee_calculator (TransactionFeeCalculator) -- TransactionFeeCalculator implementation

Return type:

Trade

Returns:

Trade that the TradeRequest represents

abstract should_execute(price)

Determine if a trade can execute given the current price.

Parameters:: price (float) -- The current price of the asset
Returns:: True if the trade can execute
Return type:: bool

class coin_test.backtest.TransactionFeeCalculator: Calculate the transactions fees for a trade.

coin_test.util

Initialize utilities for the coin-test package.

class coin_test.util.AssetPair(asset: Ticker, currency: Ticker)

Pair of tickers that can be traded.

asset: Ticker: Alias for field number 0

currency: Ticker: Alias for field number 1

static from_str(asset_str, currency_str)

Create an AssetPair from strings that represent tickers.

Return type:: AssetPair

class coin_test.util.Money(ticker, qty): Store a quantity of a given currency.

class coin_test.util.Side(value)

The side for a trade.

BUY, SELL

class coin_test.util.Ticker(symbol): Represent an asset.

class coin_test.util.TradeType(value)

The type of trade being performed.

Currently the only option is MARKET

coin_test.analysis

Analysis module of the coin-test library.

coin_test.analysis.build_datapane(results, output_dir='')

Build Datapane from large set of results.

Parameters:

results (Sequence[BacktestResults]) -- List of BacktestResults.
output_dir (str) -- Directory to save report and assets to. Defaults to local directory.

Return type:

None

Reference

coin_test.data

coin_test.backtest

coin_test.util

coin_test.analysis

Indices and tables