app.backbone.services.utils

  1from collections import namedtuple
  2from typing import List
  3import numpy as np
  4from pandas import DataFrame
  5from sklearn.linear_model import LinearRegression
  6import yaml
  7from app.backbone.entities.bot_performance import BotPerformance
  8from app.backbone.entities.trade import Trade
  9import pandas as pd
 10import uuid
 11
 12# Los ejemplos para estos test se pueden encontrar aqui:
 13# https://docs.google.com/spreadsheets/d/10OSCpBoGuY5uuCzZcLtCrbHPkMUrr1uSpLN8SDbRbH8/edit?gid=2082342944#gid=2082342944
 14
 15
 16def _performance_from_df_to_obj(
 17    df_performance: DataFrame, 
 18    date_from, 
 19    date_to, 
 20    risk, 
 21    method, 
 22    bot, 
 23    initial_cash, 
 24    ):
 25    performance_for_db = [BotPerformance(**row) for _, row in df_performance.iterrows()].pop()
 26    performance_for_db.DateFrom = date_from
 27    performance_for_db.DateTo = date_to
 28    performance_for_db.Risk = risk
 29    performance_for_db.Method = method
 30    performance_for_db.Bot = bot
 31    performance_for_db.InitialCash = initial_cash
 32    
 33    return performance_for_db
 34
 35def get_trade_df_from_db(trades: List[Trade], performance_id=None):
 36    # Obtener nombres de columnas directamente desde el modelo
 37    columns = [col.name for col in Trade.__table__.columns if col.name != 'BotPerformance']
 38
 39    if not trades:
 40        return pd.DataFrame(columns=columns)
 41
 42    # Construir los dicts con getattr dinámicamente
 43    data = [
 44        {col: getattr(trade, col) for col in columns}
 45        for trade in trades
 46    ]
 47
 48    df = pd.DataFrame(data)
 49    df['ExitTime'] = pd.to_datetime(df['ExitTime'])
 50    df = df.sort_values(by='ExitTime')
 51    df['Date'] = df['ExitTime']
 52    df.set_index('Date', inplace=True)
 53
 54    return df
 55
 56def trades_from_df_to_obj(trades: pd.DataFrame) -> List[Trade]:
 57    trade_columns = [column for column in trades.columns if hasattr(Trade, column)]
 58    trades_db = [Trade(**row[trade_columns]) for _, row in trades.iterrows()]
 59
 60    return trades_db
 61
 62def get_date_range(equity_curves: pd.DataFrame):
 63    min_date = None
 64    max_date = None
 65
 66    for name, curve in equity_curves.items():
 67
 68        if curve.empty:
 69            continue
 70        # Convertir las fechas a UTC si son tz-naive
 71        actual_date = curve.index[0].tz_localize('UTC') if curve.index[0].tz is None else curve.index[0].tz_convert('UTC')
 72        
 73        # Si min_date es None, inicializar con la primera fecha
 74        if min_date is None:
 75            min_date = actual_date
 76        # Comparar si la fecha actual es menor que min_date
 77        elif actual_date < min_date:
 78            min_date = actual_date
 79
 80        # Si max_date es None, inicializar con la última fecha
 81        curve_last_date = curve.index[-1].tz_localize('UTC') if curve.index[-1].tz is None else curve.index[-1].tz_convert('UTC')
 82        
 83        if max_date is None:
 84            max_date = curve_last_date
 85        # Comparar si la fecha actual es mayor que max_date
 86        elif curve_last_date > max_date:
 87            max_date = curve_last_date
 88
 89    # Calcular min_date y max_date
 90    min_date = min_date.date()
 91    max_date = max_date.date()
 92
 93    date_range = pd.to_datetime(pd.date_range(start=min_date, end=max_date, freq='D'))
 94    return date_range
 95
 96def calculate_stability_ratio(equity_curve: pd.Series):
 97    x = np.arange(len(equity_curve)).reshape(-1, 1)
 98    reg = LinearRegression().fit(x, equity_curve)
 99    stability_ratio = reg.score(x, equity_curve)
100    
101    return stability_ratio
102
103def max_drawdown(equity_curve, verbose=True):
104    # Calcular el running max de la equity curve
105    running_max = np.maximum.accumulate(equity_curve)
106    
107    # Calcular el drawdown
108    drawdown = (equity_curve - running_max) / running_max
109    
110    # Encontrar el valor máximo de drawdown y la fecha correspondiente
111    max_drawdown_value = np.min(drawdown) * 100  # Convertir el drawdown a porcentaje
112    max_drawdown_date = equity_curve.index[np.argmin(drawdown)]
113    
114    if verbose:
115        print(f"Máximo drawdown: {max_drawdown_value:.2f}%")
116        print(f"Fecha del máximo drawdown: {max_drawdown_date}")
117
118    return max_drawdown_value
119
120def calculate_sharpe_ratio(returns, risk_free_rate=None, trading_periods=252):
121    excess_returns = returns - (risk_free_rate / trading_periods)
122    sharpe = excess_returns.mean() / returns.std()
123    return sharpe * np.sqrt(trading_periods)