Statistics and Performance Metrics

import numpy as np
from hftbacktest.stats import LinearAssetRecord

# Load backtest results
record = np.load('backtest_results.npz')['0']  # Asset 0

# Generate statistics
stats = (
    LinearAssetRecord(record)
        .resample('10s')      # Resample to 10-second intervals
        .monthly()            # Generate monthly breakdowns
        .stats(book_size=100000)  # $100k capital
)

# Display summary
print(stats.summary())

# Plot equity curve
stats.plot()

from hftbacktest import Recorder
from numba import njit

recorder = Recorder(num_assets=1, record_size=100000)

@njit
def strategy(hbt, rec):
    while hbt.elapse(10_000_000) == 0:  # Every 10ms
        # Record current state
        rec.record(hbt)
        
        # ... strategy logic ...
    
    return True

# Run backtest
strategy(hbt, recorder.recorder)

# Save results
recorder.to_npz('results.npz')

from hftbacktest.stats import LinearAssetRecord
import numpy as np

# Load recorded data
data = np.load('backtest_results.npz')['0']

# Create record processor
record = LinearAssetRecord(data)

from hftbacktest.stats import InverseAssetRecord

record = InverseAssetRecord(data)
stats = record.resample('10s').stats(book_size=1.0)  # 1 BTC capital

from hftbacktest.stats import Metric
import polars as pl

class CustomMetric(Metric):
    def __init__(self, name='CustomMetric'):
        self.name = name
    
    def compute(self, df: pl.DataFrame, context: dict) -> dict:
        # Compute your metric
        value = df['position'].abs().mean()
        return {self.name: value}

import numpy as np
from numba import njit
from hftbacktest import (
    BacktestAsset,
    HashMapMarketDepthBacktest,
    Recorder,
    GTX, LIMIT, BUY
)
from hftbacktest.stats import (
    LinearAssetRecord,
    SR, Sortino, Ret, AnnualRet,
    MaxDrawdown, ReturnOverMDD,
    DailyNumberOfTrades,
    DailyTradingValue,
    MaxPositionValue,
    MaxLeverage
)

# Setup recorder
recorder = Recorder(num_assets=1, record_size=1_000_000)

@njit
def simple_strategy(hbt, rec):
    asset_no = 0
    
    while hbt.elapse(10_000_000) == 0:  # Every 10ms
        # Record state
        rec.record(hbt)
        
        # Simple strategy logic
        depth = hbt.depth(asset_no)
        mid = (depth.best_bid + depth.best_ask) / 2.0
        
        hbt.submit_buy_order(
            asset_no, 1, depth.best_bid,
            0.01, GTX, LIMIT, False
        )
    
    return True

# Run backtest
asset = BacktestAsset().data('data.npz').linear_asset(1.0)
hbt = HashMapMarketDepthBacktest([asset])
simple_strategy(hbt, recorder.recorder)

# Save results
recorder.to_npz('results.npz')

# Analyze performance
data = np.load('results.npz')['0']

stats = (
    LinearAssetRecord(data)
        .resample('1m')     # 1-minute resampling
        .monthly()          # Monthly breakdown
        .stats(
            metrics=[
                SR('SR365', trading_days_per_year=365),
                Sortino(),
                Ret(),
                AnnualRet(trading_days_per_year=365),
                MaxDrawdown(),
                ReturnOverMDD(),
                DailyNumberOfTrades(),
                DailyTradingValue(),
                MaxPositionValue(),
                MaxLeverage()
            ],
            book_size=100000  # $100k capital
        )
)

# Display results
print("\n=== Performance Summary ===")
summary = stats.summary()
print(summary)

# Plot equity curve
fig = stats.plot(price_as_ret=True)
fig.savefig('performance.png', dpi=300, bbox_inches='tight')
print("\nEquity curve saved to performance.png")

# Calculate required size
backtest_hours = 24 * 30  # 30 days
record_interval_seconds = 10
record_size = backtest_hours * 3600 / record_interval_seconds

recorder = Recorder(num_assets=1, record_size=int(record_size * 1.1))

stats = record.stats(book_size=100000)  # $100k capital

from hftbacktest.stats import Metric
import polars as pl

class AvgHoldingTime(Metric):
    def __init__(self):
        self.name = 'AvgHoldingTime'
    
    def compute(self, df: pl.DataFrame, context: dict) -> dict:
        # Calculate average time in position
        in_position = (df['position'] != 0).to_numpy()
        if in_position.sum() == 0:
            return {self.name: 0}
        
        # Simple approximation
        holding_ratio = in_position.mean()
        total_time = (df['timestamp'][-1] - df['timestamp'][0]).total_seconds()
        avg_time = total_time * holding_ratio
        
        return {self.name: avg_time / 3600}  # Convert to hours

stats = record.stats(metrics=[AvgHoldingTime()])