bd6a0f05d7
- Introduced `BBRSIncrementalState` for real-time processing of the Bollinger Bands + RSI strategy, allowing minute-level data input and internal timeframe aggregation. - Added `TimeframeAggregator` class to handle real-time data aggregation to higher timeframes (15min, 1h, etc.). - Updated `README_BBRS.md` to document the new incremental strategy, including key features and usage examples. - Created comprehensive tests to validate the incremental strategy against the original implementation, ensuring signal accuracy and performance consistency. - Enhanced error handling and logging for better monitoring during real-time processing. - Updated `TODO.md` to reflect the completion of the incremental BBRS strategy implementation.
112 lines
4.3 KiB
Python
112 lines
4.3 KiB
Python
"""
|
|
Debug RSI Differences
|
|
|
|
This script performs a detailed analysis of RSI calculation differences
|
|
between the original and incremental implementations.
|
|
"""
|
|
|
|
import pandas as pd
|
|
import numpy as np
|
|
import logging
|
|
from cycles.Analysis.rsi import RSI
|
|
from cycles.utils.storage import Storage
|
|
|
|
# Setup logging
|
|
logging.basicConfig(level=logging.INFO)
|
|
|
|
def debug_rsi_calculation():
|
|
"""Debug RSI calculation step by step."""
|
|
|
|
# Load small sample of data
|
|
storage = Storage(logging=logging)
|
|
data = storage.load_data("btcusd_1-min_data.csv", "2023-01-01", "2023-01-02")
|
|
|
|
# Take first 50 rows for detailed analysis
|
|
test_data = data.iloc[:50].copy()
|
|
|
|
print(f"Analyzing {len(test_data)} data points")
|
|
print(f"Price range: {test_data['close'].min():.2f} - {test_data['close'].max():.2f}")
|
|
|
|
# Original implementation
|
|
config = {"rsi_period": 14}
|
|
rsi_calculator = RSI(config=config)
|
|
original_result = rsi_calculator.calculate(test_data.copy(), price_column='close')
|
|
|
|
# Manual step-by-step calculation to understand the original
|
|
prices = test_data['close'].values
|
|
period = 14
|
|
|
|
print("\nStep-by-step manual calculation:")
|
|
print("Index | Price | Delta | Gain | Loss | AvgGain | AvgLoss | RS | RSI_Manual | RSI_Original")
|
|
print("-" * 100)
|
|
|
|
deltas = np.diff(prices)
|
|
gains = np.where(deltas > 0, deltas, 0)
|
|
losses = np.where(deltas < 0, -deltas, 0)
|
|
|
|
# Calculate using pandas EMA with Wilder's smoothing
|
|
gain_series = pd.Series(gains, index=test_data.index[1:])
|
|
loss_series = pd.Series(losses, index=test_data.index[1:])
|
|
|
|
# Wilder's smoothing: alpha = 1/period, adjust=False
|
|
avg_gain = gain_series.ewm(alpha=1/period, adjust=False, min_periods=period).mean()
|
|
avg_loss = loss_series.ewm(alpha=1/period, adjust=False, min_periods=period).mean()
|
|
|
|
rs_manual = avg_gain / avg_loss.replace(0, 1e-9)
|
|
rsi_manual = 100 - (100 / (1 + rs_manual))
|
|
|
|
# Handle edge cases
|
|
rsi_manual[avg_loss == 0] = np.where(avg_gain[avg_loss == 0] > 0, 100, 50)
|
|
rsi_manual[avg_gain.isna() | avg_loss.isna()] = np.nan
|
|
|
|
# Compare with original
|
|
for i in range(min(30, len(test_data))):
|
|
price = prices[i]
|
|
|
|
if i == 0:
|
|
print(f"{i:5d} | {price:7.2f} | - | - | - | - | - | - | - | -")
|
|
else:
|
|
delta = deltas[i-1]
|
|
gain = gains[i-1]
|
|
loss = losses[i-1]
|
|
|
|
# Get values from series (may be NaN)
|
|
avg_g = avg_gain.iloc[i-1] if i-1 < len(avg_gain) else np.nan
|
|
avg_l = avg_loss.iloc[i-1] if i-1 < len(avg_loss) else np.nan
|
|
rs_val = rs_manual.iloc[i-1] if i-1 < len(rs_manual) else np.nan
|
|
rsi_man = rsi_manual.iloc[i-1] if i-1 < len(rsi_manual) else np.nan
|
|
|
|
# Get original RSI
|
|
rsi_orig = original_result['RSI'].iloc[i] if 'RSI' in original_result.columns else np.nan
|
|
|
|
print(f"{i:5d} | {price:7.2f} | {delta:5.2f} | {gain:4.2f} | {loss:4.2f} | {avg_g:7.4f} | {avg_l:7.4f} | {rs_val:2.1f} | {rsi_man:10.4f} | {rsi_orig:10.4f}")
|
|
|
|
# Now test incremental implementation
|
|
print("\n" + "="*80)
|
|
print("INCREMENTAL IMPLEMENTATION TEST")
|
|
print("="*80)
|
|
|
|
# Test incremental
|
|
from cycles.IncStrategies.indicators.rsi import RSIState
|
|
debug_rsi = RSIState(period=14)
|
|
incremental_results = []
|
|
|
|
print("\nTesting corrected incremental RSI:")
|
|
for i, price in enumerate(prices[:20]): # First 20 values
|
|
rsi_val = debug_rsi.update(price)
|
|
incremental_results.append(rsi_val)
|
|
print(f"Step {i+1}: price={price:.2f}, RSI={rsi_val:.4f}")
|
|
|
|
print("\nComparison of first 20 values:")
|
|
print("Index | Original RSI | Incremental RSI | Difference")
|
|
print("-" * 50)
|
|
|
|
for i in range(min(20, len(original_result))):
|
|
orig_rsi = original_result['RSI'].iloc[i] if 'RSI' in original_result.columns else np.nan
|
|
inc_rsi = incremental_results[i] if i < len(incremental_results) else np.nan
|
|
diff = abs(orig_rsi - inc_rsi) if not (np.isnan(orig_rsi) or np.isnan(inc_rsi)) else np.nan
|
|
|
|
print(f"{i:5d} | {orig_rsi:11.4f} | {inc_rsi:14.4f} | {diff:10.4f}")
|
|
|
|
if __name__ == "__main__":
|
|
debug_rsi_calculation() |