Ridge Classification of Crypto Futures Direction from Other Futures Prices

# strategy-cryptofutures-ml-ridge-with-futures


# strategy-cryptofutures-ml-ridge-with-futures









## Source (MIT)

```python
# # Machine learning - RidgeClassifier (log futures close prices)
# 
# Your user account:
# 
# * [User account](https://quantiacs.io/personalpage/homepage)
# 
# Documentation:
# 
# * [Documentation](https://quantiacs.io/documentation/en/)
# 
# 
# **Strategy idea:** We will open cryptofutures positions as predicted by the **RidgeClassifier**.
# 
# **Features for learning** - the logarithm of closing prices for the last 18 days of the futures **"F_O", "F_LN", "F_KC"**

%%javascript
IPython.OutputArea.prototype._should_scroll = function(lines) { return false; }
// disable widget scrolling

import xarray as xr

import qnt.backtester as qnbt
import qnt.data as qndata
import numpy as np
import pandas as pd


def load_data(period):
    futures = qndata.futures.load_data(tail=period, assets=["F_O", "F_LN", "F_KC"])
    crypto = qndata.cryptofutures.load_data(tail=period)
    return {"futures": futures, "crypto": crypto}, futures.time.values


def build_data_for_one_step(data, max_date: np.datetime64, lookback_period: int):
    min_date = max_date - np.timedelta64(lookback_period, "D")
    return {
        "futures": data["futures"].sel(time=slice(min_date, max_date)),
        "crypto": data["crypto"].sel(time=slice(min_date, max_date)),
    }


def predict_weights(market_data):
    def get_ml_model():
        from sklearn.linear_model import RidgeClassifier
        model = RidgeClassifier(random_state=18)
        return model

    def get_features(data):
        def remove_trend(prices_pandas_):
            prices_pandas = prices_pandas_.copy(True)
            assets = prices_pandas.columns
            for asset in assets:
                prices_pandas[asset] = np.log(prices_pandas[asset])
            return prices_pandas

        price = data.sel(field="close").ffill('time').bfill('time').fillna(0) # fill NaN
        for_result = price.to_pandas()
        features_no_trend_df = remove_trend(for_result)
        return features_no_trend_df

    def get_target_classes(data):

        price_current = data.sel(field="close").dropna('time')
        price_future = price_current.shift(time=-1).dropna('time')

        class_positive = 1
        class_negative = 0

        target_is_price_up = xr.where(price_future > price_current, class_positive, class_negative)
        return target_is_price_up.to_pandas()

    futures = market_data["futures"].copy(True)
    crypto = market_data["crypto"].copy(True)

    asset_name_all = crypto.coords['asset'].values
    features_all_df = get_features(futures)
    target_all_df = get_target_classes(crypto)

    predict_weights_next_day_df = crypto.sel(field="close").isel(time=-1).to_pandas()

    for asset_name in asset_name_all:
        target_for_learn_df = target_all_df[asset_name]
        feature_for_learn_df = features_all_df[:-1] # last value reserved for prediction

        # align features and targets
        target_for_learn_df, feature_for_learn_df = target_for_learn_df.align(feature_for_learn_df, axis=0, join='inner')

        model = get_ml_model()

        try:
            model.fit(feature_for_learn_df.values, target_for_learn_df)

            feature_for_predict_df = features_all_df[-1:]

            predict = model.predict(feature_for_predict_df.values)
            predict_weights_next_day_df[asset_name] = predict
        except:
            logging.exception("model failed")
            # if there is exception, return zero values
            return xr.zeros_like(crypto.isel(field=0, time=0))


    return predict_weights_next_day_df.to_xarray()


weights = qnbt.backtest(
    competition_type="cryptofutures",
    load_data=load_data,
    lookback_period=18,
    start_date='2014-01-01',
    strategy=predict_weights,
    window=build_data_for_one_step,
    analyze=True,
    build_plots=True
)

# # What libraries are available?
# 
# Our library makes extensive use of xarray: 
# 
# [xarray](http://xarray.pydata.org/en/stable/)
# 
# pandas:
# 
# [pandas](https://pandas.pydata.org/)
# 
# and numpy:
# 
# [numpy](https://numpy.org/)
# 
# Function definitions can be found in the qnt folder in your private root directory.
# 
# ```python
# # Import basic libraries.
# import xarray as xr
# import pandas as pd
# import numpy as np
# 
# # Import quantnet libraries.
# import qnt.data    as qndata  # load and manipulate data
# import qnt.output as output   # manage output
# import qnt.backtester as qnbt # backtester
# import qnt.stats   as qnstats # statistical functions for analysis
# import qnt.graph   as qngraph # graphical tools
# import qnt.ta      as qnta    # indicators library
# ```

# # May I import libraries?
# 
# Yes, please refer to the file **init.ipynb** in your home directory. You can dor example use:
# 
# ! conda install -y scikit-learn

# # How to load data?
# 
# Futures:
# ```python
# data= qndata.futures.load_data(tail = 15*365, dims = ("time", "field", "asset"))
# ```
# 
# BTC Futures:
# ```python
# data= qndata.cryptofutures.load_data(tail = 15*365, dims = ("time", "field", "asset"))
# ```
# 
# Cryptocurrencies:
# ```python
# data= qndata.crypto.load_data(tail = 15*365, dims = ("time", "field", "asset"))
# ```

# # How to view a list of all tickers?
# 
# ```python
# data.asset.to_pandas().to_list()
# ```

# # How to see which fields are available?
# 
# ```python
# data.field.to_pandas().to_list()
# ```

# # How to load specific tickers?
# 
# ```python
# data = qndata.futures.load_data(tail=15 * 365, assets=['F_O', 'F_DX', 'F_GC'])
# ```

# # How to select specific tickers after loading all data?
# 
# ```python
# def get_data_filter(data, assets):
#     filler= data.sel(asset=assets)
#     return filler
# 
# get_data_filter(data, ["F_O", "F_DX", "F_GC"])
# ```

# # How to get the prices for the previous day?
# 
# ```python
# qnta.shift(data.sel(field="open"), periods=1)
# ```
# 
# or:
# 
# ```python
# data.sel(field="open").shift(time=1)
# ```

# # How do I get a list of the top 10 assets ranked by Sharpe ratio?
# 
# ```python
# import qnt.stats as qnstats
# 
# data= qndata.futures.load_data(tail=16 * 365)
# 
# def get_best_instruments(data, weights, top_size):
#     # compute statistics:
#     stats_per_asset= qnstats.calc_stat(data, weights, per_asset=True)
#     # calculate ranks of assets by "sharpe_ratio":
#     ranks= (-stats_per_asset.sel(field="sharpe_ratio")).rank("asset")
#     # select top assets by rank "top_period" days ago:
#     top_period= 300
#     rank= ranks.isel(time=-top_period)
#     top= rank.where(rank <= top_size).dropna("asset").asset
# 
#     # select top stats:
#     top_stats= stats_per_asset.sel(asset=top.values)
# 
#     # print results:
#     print("SR tail of the top assets:")
#     display(top_stats.sel(field="sharpe_ratio").to_pandas().tail())
# 
#     print("avg SR = ", top_stats[-top_period:].sel(field="sharpe_ratio").mean("asset")[-1].item())
#     display(top_stats)
#     return top_stats.coords["asset"].values
# 
# get_best_instruments(data, weights, 10)
# ```

# # How can I check the results for only the top 10 assets ranked by Sharpe ratio?
# 
# Select the top assets and then load their data:
# 
# ```python
# best_assets= get_best_instruments(data, weights, 10)
# 
# data= qndata.futures.load_data(tail=15 * 365, assets=best_assets)
# ...
# ```

# # How can prices be processed?
# 
# Simply import standard libraries, for example **numpy**:
# 
# ```python
# import numpy as np
# 
# high= np.log(data.sel(field="high"))
# ```

# # How can you reduce slippage impace when trading?
# 
# Just apply some technique to reduce turnover:
# 
# ```python
# def get_lower_slippage(weights, rolling_time=6):
#     return weights.rolling({"time": rolling_time}).max()
# 
# improved_weights = get_lower_slippage(weights, rolling_time=6)
# ```

# # How to use technical analysis indicators?
# 
# For available indicators see the source code of the library: /qnt/ta
# 
# ## ATR
# 
# ```python
# def get_atr(data, days=14):
#     high = data.sel(field="high") * 1.0 
#     low  = data.sel(field="low") * 1.0 
#     close= data.sel(field="close") * 1.0
# 
#     return qnta.atr(high, low, close, days)
# 
# atr= get_atr(data, days=14)
# ```
# 
# ## EMA
# 
# ```python
# prices= data.sel(field="high")
# prices_ema= qnta.ema(prices, 15)
# ```
# 
# ## TRIX
# 
# ```python
# prices= data.sel(field="high")
# prices_trix= qnta.trix(prices, 15)
# ```
# 
# ## ADL and EMA
# 
# ```python
# adl= qnta.ad_line(data.sel(field="close")) * 1.0 
# adl_ema= qnta.ema(adl, 18)
# ```

# # How can you check the quality of your strategy?
# 
# ```python
# import qnt.output as qnout
# qnout.check(weights, data)
# ```
# 
# or
# 
# ```python
# stat= qnstats.calc_stat(data, weights)
# display(stat.to_pandas().tail())
# ```
# 
# or
# 
# ```python
# import qnt.graph   as qngraph
# statistics= qnstats.calc_stat(data, weights)
# display(statistics.to_pandas().tail())
# 
# performance= statistics.to_pandas()["equity"]
# qngraph.make_plot_filled(performance.index, performance, name="PnL (Equity)", type="log")
# 
# display(statistics[-1:].sel(field = ["sharpe_ratio"]).transpose().to_pandas())
# qnstats.print_correlation(weights, data)
# 
# ```

# # An example using pandas
# 
# One can work with pandas DataFrames at intermediate steps and at the end convert them to xarray data structures:
# 
# ```python
# def get_price_pct_change(prices):
#     prices_pandas= prices.to_pandas()
#     assets= data.coords["asset"].values
#     for asset in assets:
#         prices_pandas[asset]= prices_pandas[asset].pct_change()
#     return prices_pandas
# 
# 
# prices= data.sel(field="close") * 1.0
# prices_pct_change= get_price_pct_change(prices).unstack().to_xarray()
# 
# ```

# # How to submit a strategy to the competition?
# 
# Check that weights are fine:
# 
# ```python
# import qnt.output as qnout
# qnout.check(weights, data)
# ```
# 
# If everything is ok, write the weights to file:
# 
# ```python
# qnout.write(weights)
# ```
# 
# In your **personal account**:
# 
# * **choose** a strategy;
# * click on the **Submit** button;
# * select the type of competition.
# 
# At the beginning you will find the strategy under the **Checking** area (**Competition** > **Checking**). If Sharpe ratio is larger than 1 and technical checks are successful, the strategy will go under the **Running** area (**Competition** > **Running**). Otherwise it will be **Filtered** (**Competition** > **Filtered**) and you should inspect error and warning messages.

```