moving_average_python_enhanced.py

import numpy as np
import pandas as pd
from numba import jit
import time


@jit(nopython=True)
def calculate_moving_average_numba(prices, period):
    """Calculate moving average using Numba for JIT compilation"""
    ma = np.zeros(len(prices) - period + 1)
    window_sum = np.sum(prices[:period])
    ma[0] = window_sum / period
    
    for i in range(1, len(ma)):
        window_sum = window_sum - prices[i - 1] + prices[i + period - 1]
        ma[i] = window_sum / period
    
    return ma


@jit(nopython=True)
def calculate_complex_math_numba(opens, highs, lows, closes):
    """Perform complex mathematical operations using Numba"""
    results = np.zeros(len(closes))
    for i in range(len(closes)):
        # Complex mathematical function: weighted combination of OHLC values
        value = (opens[i] * 0.1) + (highs[i] * 0.3) + (lows[i] * 0.2) + (closes[i] * 0.4)
        # Apply exponential and logarithmic transformations
        results[i] = np.exp(value / 100.0) * np.log(np.abs(value) + 1.0)
    return results


def calculate_multiple_moving_averages_numba(prices, periods):
    """Calculate multiple moving averages for different periods"""
    results = {}
    for period in periods:
        if len(prices) >= period:
            results[f'MA_{period}'] = calculate_moving_average_numba(prices, period)
    return results


def main():
    start_time = time.time()
    
    print("Reading CSV file with pandas...")
    df = pd.read_csv("USDJPY2.csv")
    print(f"Loaded {len(df)} records.")
    
    # Extract price arrays
    opens = df['Open'].values.astype(np.float64)
    highs = df['High'].values.astype(np.float64)
    lows = df['Low'].values.astype(np.float64)
    closes = df['Close'].values.astype(np.float64)
    
    # Calculate multiple moving averages for periods 200-220
    print("Calculating moving averages for periods 200-220...")
    ma_periods = list(range(200, 221))  # 200 to 220 inclusive
    all_mas = calculate_multiple_moving_averages_numba(closes, ma_periods)
    
    for period, ma_values in all_mas.items():
        print(f"Calculated {len(ma_values)} {period} values.")
    
    # Perform complex mathematical operations
    print("Performing complex mathematical operations...")
    complex_results = calculate_complex_math_numba(opens, highs, lows, closes)
    print(f"Completed {len(complex_results)} complex calculations.")
    
    end_time = time.time()
    duration = (end_time - start_time) * 1000  # Convert to milliseconds
    print(f"Total execution time: {duration:.2f} ms")


if __name__ == "__main__":
    main()