jan2026_reasons_for_trend.py

import datetime
import matplotlib.patches as mpatches
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

from syscore.dateutils import SECONDS_IN_YEAR
from systems.provided.rob_system.run_system import futures_system
from private.projects.futures_book.generate_instrument_list import MASTER_INSTRUMENT_LIST


def calculate_synthetic_spot(system, instrument_code):
    price = system.data.daily_prices(instrument_code)

    cum_carry = calculate_cum_carry(system, instrument_code)
    syn_spot = price - cum_carry

    return syn_spot


def calculate_cum_carry(system, instrument_code):
    price = system.data.daily_prices(instrument_code)
    ann_carry = system.rawdata.daily_annualised_roll(instrument_code)
    ann_carry_fitted = ann_carry.reindex(price.index, method="ffill")
    diff_index_in_years_as_pd = pd_series_of_diff_index_in_years(ann_carry_fitted)

    carry_per_period = diff_index_in_years_as_pd * ann_carry_fitted
    cum_carry = carry_per_period.cumsum()

    return cum_carry

def pd_series_of_diff_index_in_years(x: pd.Series):
    diff_index_in_years = get_annual_intervals_from_series(x)

    return pd.Series([0] + diff_index_in_years, x.index)


def get_annual_intervals_from_series(x: pd.Series):
    diff_index = x[1:].index - x[:-1].index
    diff_index_as_list = list(diff_index)
    diff_index_in_seconds = [
        index_item.total_seconds() for index_item in diff_index_as_list
    ]
    diff_index_in_years = [
        index_item_in_seconds / SECONDS_IN_YEAR
        for index_item_in_seconds in diff_index_in_seconds
    ]

    return diff_index_in_years



system = futures_system()

adjusted_price_dict = dict([(key,
                    system.rawdata.get_daily_prices(key))
                    for key in MASTER_INSTRUMENT_LIST
     ]
)
adjusted_price_df = pd.DataFrame(adjusted_price_dict)
adjusted_price_df = adjusted_price_df.resample("1B").last()

carry_price_dict =  dict([(key,
                    calculate_cum_carry(system, key))
                    for key in MASTER_INSTRUMENT_LIST
     ]
)
carry_price_df = pd.DataFrame(carry_price_dict)
carry_price_df = carry_price_df.resample("1B").last()

synthetic_price_dict =  dict([(key,
                    calculate_synthetic_spot(system, key))
                    for key in MASTER_INSTRUMENT_LIST
     ]
)
synthetic_price_df = pd.DataFrame(synthetic_price_dict)
synthetic_price_df = synthetic_price_df.resample("1B").last()

strategy_return_dict = dict(
    [(rule_name,
      dict(
          [
              (instrument, system.accounts.pandl_for_instrument_forecast(instrument, rule_name ).percent)
              for instrument in MASTER_INSTRUMENT_LIST
          ]
      )
     ) for rule_name in ['momentum8', 'momentum16', 'momentum64']]
)

from sklearn.linear_model import LinearRegression

def do_a_plot(rule_name: str, underlying_price: pd.DataFrame, price_label: str, scatter=True, regr_line=False):
    list_of_dict =  get_a_list_of_tuples_underlying_and_strat_SR_for_five_year_blocks(rule_name, underlying_price)
    dict_as_df = pd.DataFrame(list_of_dict)
    asset_classes = list(set(dict_as_df.asset.values))
    colour_list = ['red', 'black', 'yellow', 'green', 'blue', 'gray', 'orange']
    patchList = []
    for asset, color in zip(asset_classes, colour_list):
        data_key = mpatches.Patch(color=color, label=asset)
        patchList.append(data_key)
        this_asset = dict_as_df[dict_as_df.asset == asset]
        x = this_asset['price_SR'].values
        y = this_asset['trend_SR'].values
        X = x.reshape(-1, 1)
        Y = y.reshape(-1, 1)
        linear_regressor = LinearRegression()
        linear_regressor.fit(X, Y)
        Y_pred = linear_regressor.predict(X)
        if scatter:
            plt.scatter(x, y, color=color)
        if regr_line:
            plt.plot(X, Y_pred, color=color)
    plt.title("%s %s" % (price_label, rule_name))
    plt.xlabel('Abs(SR) of %s' % price_label)
    plt.ylabel("SR of %s" % rule_name)
    plt.legend(handles=patchList)
    plt.show(block=True)

def get_distribution_of_SR( underlying_price: pd.DataFrame, price_label: str):
    arbitrary_rule_name = "momentum8"
    list_of_dict =  get_a_list_of_tuples_underlying_and_strat_SR_for_five_year_blocks(arbitrary_rule_name, underlying_price)
    dict_as_df = pd.DataFrame(list_of_dict)
    asset_classes = list(set(dict_as_df.asset.values))
    new_dict={}
    for asset in asset_classes:
        this_asset = dict_as_df[dict_as_df.asset == asset]
        new_dict[asset] = pd.Series(this_asset.price_SR.values, name=asset)
    new_df = pd.concat(new_dict, axis=1)
    new_df.boxplot()
    plt.title("%s returns" % (price_label))
    plt.show(block=True)

    return new_df.median(axis=0)

def get_distribution_of_SR_rule(rule_name: str):
    arbitrary_underlying_price = adjusted_price_df
    list_of_dict =  get_a_list_of_tuples_underlying_and_strat_SR_for_five_year_blocks(rule_name, arbitrary_underlying_price)
    dict_as_df = pd.DataFrame(list_of_dict)
    asset_classes = list(set(dict_as_df.asset.values))
    new_dict={}
    for asset in asset_classes:
        this_asset = dict_as_df[dict_as_df.asset == asset]
        new_dict[asset] = pd.Series(this_asset.trend_SR.values, name=asset)
    new_df = pd.concat(new_dict, axis=1)
    new_df.boxplot()
    plt.title("%s returns" % (rule_name))
    plt.show(block=True)

    return new_df.median(axis=0)


def get_a_list_of_tuples_underlying_and_strat_SR_for_five_year_blocks(rule_name: str, underlying_price: pd.DataFrame):
    list_of_starts =pd.date_range(underlying_price.index[0], end=underlying_price.index[-1], freq="60M")
    list_of_ends = list(list_of_starts[1:])
    list_of_starts =list(list_of_starts[:-1])
    list_of_tuples = []
    for start, end in zip(list_of_starts, list_of_ends):
        print(start)
        list_of_tuples+=        get_tuples_underlying_and_strat_SR_for_five_year_block(
            rule_name=rule_name,
            underlying_price=underlying_price,
            start=start,end=end
        )

    return list_of_tuples


def get_tuples_underlying_and_strat_SR_for_five_year_block(rule_name: str, underlying_price: pd.DataFrame, start:datetime.datetime, end:datetime.datetime):
    list_of_tuples = []
    for instrument in MASTER_INSTRUMENT_LIST:
        tups =get_tuples_underlying_and_strat_SR_for_five_year_block_one_instrument(
            rule_name=rule_name,
            underlying_price=underlying_price,
            start=start, end=end,
            instrument=instrument
        )
        if np.isnan(tups['price_SR']) or np.isnan(tups['trend_SR']):
            continue

        list_of_tuples.append(tups)

    return list_of_tuples

def get_tuples_underlying_and_strat_SR_for_five_year_block_one_instrument(instrument: str,
                                                                          rule_name: str, underlying_price: pd.DataFrame, start:datetime.datetime, end:datetime.datetime):

    asset_class = system.data.asset_class_for_instrument(instrument)
    if asset_class=="Sector":
        asset_class="Equity"
    x=get_SR_for_price_series_in_time_period(
            underlying_price=underlying_price,
            start=start, end=end,
            instrument=instrument)
    y=get_SR_for_trend_rule_in_time_period( rule_name=rule_name,
            start=start, end=end,
            instrument=instrument)

    return dict(asset=asset_class, price_SR=x, trend_SR=y)

def get_SR_for_price_series_in_time_period(instrument: str,
                                                                         underlying_price: pd.DataFrame, start:datetime.datetime, end:datetime.datetime):
    px = underlying_price[instrument][start:end]
    returns = px.diff()

    return abs(16*returns.mean()/returns.std())

def get_SR_for_trend_rule_in_time_period(instrument: str,
                                                                          rule_name: str, start:datetime.datetime, end:datetime.datetime):
    returns = strategy_return_dict[rule_name][instrument][start:end]

    SR= 16*returns.mean()/returns.std()

    return SR

import matplotlib
matplotlib.rcParams.update({"font.size": 16})
rule_name="momentum64"
do_a_plot(rule_name, adjusted_price_df, "adjusted prices")
do_a_plot(rule_name, synthetic_price_df, "spot prices")
do_a_plot(rule_name, carry_price_df, "cumulative carry")


do_a_plot(rule_name, adjusted_price_df, "adjusted prices", scatter=False,regr_line=True)
do_a_plot(rule_name, synthetic_price_df, "spot prices", scatter=False,regr_line=True)
do_a_plot(rule_name, carry_price_df, "cumulative carry", scatter=False,regr_line=True)

x=get_distribution_of_SR(adjusted_price_df, "adjusted prices")
y=get_distribution_of_SR_rule(rule_name)

new_df=pd.DataFrame([x,y], index=['prices','trend'])
new_df = new_df.transpose()

fig, axs = plt.subplots()
axs.scatter(data=new_df, x='prices', y='trend')

for i, row in new_df.iterrows():
    axs.annotate( row.name, (row["prices"], row["trend"]))
plt.xlabel("Adjusted price SR")
plt.ylabel("%s SR" % rule_name)