forex-scalping.com

Code: Select all

import pandas as pd
import numpy as np

def evaluate_trade_history(csv_path, starting_balance, profit_column='Profit', minimum_acceptable_return=0.0):
    """
    Calculates Sortino and Omega ratios from a raw trading export.
    
    csv_path: Path to the exported CSV (e.g., from MT4/MT5 or cTrader).
    starting_balance: The initial account balance before the first trade in the series.
    profit_column: The name of the column containing the net PnL (after commissions/swaps).
    minimum_acceptable_return (MAR): The threshold for success (usually 0.0 for raw returns).
    """
    # Load the trading data
    try:
        df = pd.read_csv(csv_path)
    except FileNotFoundError:
        return "Error: CSV file not found."

    if profit_column not in df.columns:
        return f"Error: Column '{profit_column}' not found. Please verify your CSV headers."

    # 1. Reconstruct the equity curve to find percentage returns per trade
    # Shift by 1 so the balance reflects the capital available *before* the trade executed
    df['Running_Balance'] = starting_balance + df[profit_column].cumsum().shift(1).fillna(0)
    
    # Calculate fractional return per trade
    df['Return'] = df[profit_column] / df['Running_Balance']

    # 2. Sortino Ratio Calculation
    # Isolate downside returns (trades that fell below the minimum acceptable return)
    downside_returns = df.loc[df['Return'] < minimum_acceptable_return, 'Return']
    
    # Calculate downside deviation (Root Mean Square of downside returns)
    if len(downside_returns) > 0:
        downside_dev = np.sqrt(np.mean((downside_returns - minimum_acceptable_return)**2))
    else:
        downside_dev = 0.0

    mean_return = df['Return'].mean()
    
    # Calculate per-trade Sortino. 
    # Note: To annualize this for a 2-6 week series, multiply by np.sqrt(expected_trades_per_year)
    if downside_dev > 0:
        sortino = (mean_return - minimum_acceptable_return) / downside_dev
    else:
        sortino = np.inf  # Indicates no downside volatility

    # 3. Omega Ratio Calculation
    # Sum of all returns above the MAR
    returns_above_mar = df.loc[df['Return'] > minimum_acceptable_return, 'Return'] - minimum_acceptable_return
    sum_positive = returns_above_mar.sum()

    # Sum of all absolute losses below the MAR
    returns_below_mar = minimum_acceptable_return - df.loc[df['Return'] < minimum_acceptable_return, 'Return']
    sum_negative = returns_below_mar.sum()

    if sum_negative > 0:
        omega = sum_positive / sum_negative
    else:
        omega = np.inf  # Indicates no losses below the threshold

    return {
        "Total Executed Trades": len(df),
        "Starting Balance": starting_balance,
        "Ending Balance": round(df['Running_Balance'].iloc[-1] + df[profit_column].iloc[-1], 2),
        "Sortino Ratio (Per Trade)": round(sortino, 4),
        "Omega Ratio": round(omega, 4)
    }

# --- Execution Example ---
# Save your export as 'history.csv' in the same directory.
# Adjust the 'profit_column' name to exactly match your CSV header.
#
# results = evaluate_trade_history(
#     csv_path='history.csv', 
#     starting_balance=1000.00, 
#     profit_column='Profit' 
# )
#
# for metric, value in results.items():
#     print(f"{metric}: {value}")

Code: Select all

import pandas as pd
import numpy as np

def evaluate_trade_history(csv_path, starting_balance, profit_column='Profit', time_column='Time', minimum_acceptable_return=0.0):
    """
    Calculates Sortino, Omega, Max Drawdown, Absolute Drawdown, CAGR, and the MAR Ratio.
    
    csv_path: Path to the exported CSV.
    starting_balance: The initial account balance before the first trade.
    profit_column: Column name containing net PnL (profit + swap + commission).
    time_column: Column name containing the execution or close timestamp.
    """
    try:
        df = pd.read_csv(csv_path)
    except FileNotFoundError:
        return "Error: CSV file not found."

    if profit_column not in df.columns or time_column not in df.columns:
        return f"Error: Missing required columns. Ensure '{profit_column}' and '{time_column}' exist."

    # Parse dates and ensure chronological order
    df[time_column] = pd.to_datetime(df[time_column])
    df = df.sort_values(by=time_column).reset_index(drop=True)

    # 1. Equity Curve Reconstruction
    df['Running_Balance'] = starting_balance + df[profit_column].cumsum().shift(1).fillna(0)
    df['Equity'] = df['Running_Balance'] + df[profit_column]
    df['Return'] = df[profit_column] / df['Running_Balance']

    # 2. Drawdown Calculations
    # Peak Equity tracks the highest water mark up to the current trade
    df['Peak_Equity'] = df['Equity'].cummax()
    df['Peak_Equity'] = df['Peak_Equity'].clip(lower=starting_balance)
    
    # Maximum Peak-to-Trough Drawdown (Percentage)
    df['Drawdown_Pct'] = (df['Peak_Equity'] - df['Equity']) / df['Peak_Equity']
    max_drawdown = df['Drawdown_Pct'].max()
    
    # Absolute Drawdown (Maximum drop below the initial starting balance)
    min_equity = df['Equity'].min()
    absolute_drawdown = max(0.0, (starting_balance - min_equity) / starting_balance)

    # 3. CAGR Calculation
    # Calculate exact duration in years based on the first and last trade timestamps
    total_time_days = (df[time_column].iloc[-1] - df[time_column].iloc[0]).total_seconds() / (24 * 3600)
    years = total_time_days / 365.25
    ending_balance = df['Equity'].iloc[-1]

    if years > 0:
        cagr = (ending_balance / starting_balance) ** (1 / years) - 1
    else:
        cagr = (ending_balance - starting_balance) / starting_balance # Fallback if history < 1 day

    # 4. MAR Ratio Calculation
    if max_drawdown > 0:
        mar_ratio = cagr / max_drawdown
    else:
        mar_ratio = np.inf # No drawdown occurred

    # 5. Sortino & Omega (Retained from previous)
    downside_returns = df.loc[df['Return'] < minimum_acceptable_return, 'Return']
    downside_dev = np.sqrt(np.mean((downside_returns - minimum_acceptable_return)**2)) if len(downside_returns) > 0 else 0.0
    
    mean_return = df['Return'].mean()
    sortino = (mean_return - minimum_acceptable_return) / downside_dev if downside_dev > 0 else np.inf

    returns_above_mar = df.loc[df['Return'] > minimum_acceptable_return, 'Return'] - minimum_acceptable_return
    sum_positive = returns_above_mar.sum()
    returns_below_mar = minimum_acceptable_return - df.loc[df['Return'] < minimum_acceptable_return, 'Return']
    sum_negative = returns_below_mar.sum()
    omega = sum_positive / sum_negative if sum_negative > 0 else np.inf

    return {
        "Total Executed Trades": len(df),
        "Duration (Days)": round(total_time_days, 2),
        "Starting Balance": starting_balance,
        "Ending Balance": round(ending_balance, 2),
        "Absolute Drawdown": f"{round(absolute_drawdown * 100, 2)}%",
        "Max Drawdown (Peak-to-Trough)": f"{round(max_drawdown * 100, 2)}%",
        "CAGR": f"{round(cagr * 100, 2)}%",
        "MAR Ratio": round(mar_ratio, 4),
        "Sortino Ratio (Per Trade)": round(sortino, 4),
        "Omega Ratio": round(omega, 4)
    }

# --- Execution Example ---
# results = evaluate_trade_history(
#     csv_path='history.csv', 
#     starting_balance=1000.00, 
#     profit_column='Profit',
#     time_column='Time'
# )

Code: Select all

import os
import glob
import pandas as pd
import numpy as np

def evaluate_annual_series(export_directory, starting_balance, profit_column='Profit', time_column='Time', minimum_acceptable_return=0.0):
    """
    Ingests all CSV exports in a directory, stitches them chronologically, 
    and calculates annual metrics including MAR, Sortino, and Omega.
    
    export_directory: Path to the folder containing your series CSVs (e.g., './2026_exports/').
    """
    # 1. Find and load all CSVs in the target directory
    csv_pattern = os.path.join(export_directory, "*.csv")
    all_files = glob.glob(csv_pattern)
    
    if not all_files:
        return f"Error: No CSV files found in directory '{export_directory}'."
    
    df_list = []
    for file in all_files:
        try:
            temp_df = pd.read_csv(file)
            # Validate required columns exist before appending
            if profit_column in temp_df.columns and time_column in temp_df.columns:
                df_list.append(temp_df)
            else:
                print(f"Warning: Skipping {file} - Missing '{profit_column}' or '{time_column}'")
        except Exception as e:
            print(f"Error reading {file}: {e}")
            
    if not df_list:
        return "Error: No valid data could be extracted from the CSVs."

    # 2. Combine, clean, and chronologically sort the dataset
    df = pd.concat(df_list, ignore_index=True)
    
    # Parse dates, drop overlapping duplicates (if exports overlapped), and sort
    df[time_column] = pd.to_datetime(df[time_column])
    df = df.drop_duplicates()
    df = df.sort_values(by=time_column).reset_index(drop=True)

    # 3. Continuous Equity Curve Reconstruction
    # NOTE: This assumes profits roll over continuously. 
    df['Running_Balance'] = starting_balance + df[profit_column].cumsum().shift(1).fillna(0)
    df['Equity'] = df['Running_Balance'] + df[profit_column]
    df['Return'] = df[profit_column] / df['Running_Balance']

    # 4. Drawdown Calculations
    df['Peak_Equity'] = df['Equity'].cummax()
    df['Peak_Equity'] = df['Peak_Equity'].clip(lower=starting_balance)
    
    df['Drawdown_Pct'] = (df['Peak_Equity'] - df['Equity']) / df['Peak_Equity']
    max_drawdown = df['Drawdown_Pct'].max()
    
    min_equity = df['Equity'].min()
    absolute_drawdown = max(0.0, (starting_balance - min_equity) / starting_balance)

    # 5. Annualized CAGR Calculation
    total_time_days = (df[time_column].iloc[-1] - df[time_column].iloc[0]).total_seconds() / (24 * 3600)
    years = total_time_days / 365.25
    ending_balance = df['Equity'].iloc[-1]

    if years > 0:
        cagr = (ending_balance / starting_balance) ** (1 / years) - 1
    else:
        cagr = (ending_balance - starting_balance) / starting_balance

    # 6. MAR Ratio
    mar_ratio = cagr / max_drawdown if max_drawdown > 0 else np.inf

    # 7. Sortino & Omega Ratios
    downside_returns = df.loc[df['Return'] < minimum_acceptable_return, 'Return']
    downside_dev = np.sqrt(np.mean((downside_returns - minimum_acceptable_return)**2)) if len(downside_returns) > 0 else 0.0
    
    mean_return = df['Return'].mean()
    sortino = (mean_return - minimum_acceptable_return) / downside_dev if downside_dev > 0 else np.inf

    returns_above_mar = df.loc[df['Return'] > minimum_acceptable_return, 'Return'] - minimum_acceptable_return
    sum_positive = returns_above_mar.sum()
    returns_below_mar = minimum_acceptable_return - df.loc[df['Return'] < minimum_acceptable_return, 'Return']
    sum_negative = returns_below_mar.sum()
    omega = sum_positive / sum_negative if sum_negative > 0 else np.inf

    return {
        "Total Series CSVs Processed": len(all_files),
        "Total Executed Trades": len(df),
        "Total Calendar Days (Inc. Vacations)": round(total_time_days, 1),
        "Starting Balance": starting_balance,
        "Ending Balance": round(ending_balance, 2),
        "Absolute Drawdown": f"{round(absolute_drawdown * 100, 2)}%",
        "Max Drawdown (Peak-to-Trough)": f"{round(max_drawdown * 100, 2)}%",
        "CAGR": f"{round(cagr * 100, 2)}%",
        "MAR Ratio": round(mar_ratio, 4),
        "Sortino Ratio (Per Trade)": round(sortino, 4),
        "Omega Ratio": round(omega, 4)
    }

# --- Execution Example ---
# Place all your 2-6 week series CSVs into a folder named "2026_trading_blocks"
#
# results = evaluate_annual_series(
#     export_directory='./2026_trading_blocks/', 
#     starting_balance=1000.00, 
#     profit_column='Profit',
#     time_column='Time'
# )