moving_average_python.py

import csv
import time
from collections import deque
from math import exp, log


class Candlestick:
    def __init__(self, timestamp, open_price, high, low, close, volume):
        self.timestamp = timestamp
        self.open = float(open_price)
        self.high = float(high)
        self.low = float(low)
        self.close = float(close)
        self.volume = float(volume)


def read_csv(filename):
    data = []
    with open(filename, 'r') as file:
        reader = csv.reader(file)
        next(reader)  # Skip header
        
        for row in reader:
            if len(row) >= 6:
                candle = Candlestick(
                    row[0], row[1], row[2], row[3], row[4], row[5]
                )
                data.append(candle)
    return data


def calculate_moving_average(data, period):
    ma = []
    if len(data) < period:
        return ma
    
    # Calculate initial sum for the first window
    window_sum = sum(candle.close for candle in data[:period])
    ma.append(window_sum / period)
    
    # Use sliding window technique for efficiency
    for i in range(period, len(data)):
        window_sum = window_sum - data[i - period].close + data[i].close
        ma.append(window_sum / period)
    
    return ma


def calculate_complex_math(data):
    results = []
    for candle in data:
        # Complex mathematical function: weighted combination of OHLC values
        value = (candle.open * 0.1) + (candle.high * 0.3) + (candle.low * 0.2) + (candle.close * 0.4)
        # Apply exponential and logarithmic transformations
        transformed_value = exp(value / 100.0) * log(abs(value) + 1.0)
        results.append(transformed_value)
    
    return results


def main():
    start_time = time.time()
    
    print("Reading CSV file...")
    data = read_csv("USDJPY2.csv")
    print(f"Loaded {len(data)} records.")
    
    # Calculate multiple moving averages
    print("Calculating 50-period MA...")
    ma50 = calculate_moving_average(data, 50)
    print(f"Calculated {len(ma50)} MA50 values.")
    
    print("Calculating 200-period MA...")
    ma200 = calculate_moving_average(data, 200)
    print(f"Calculated {len(ma200)} MA200 values.")
    
    print("Calculating 500-period MA...")
    ma500 = calculate_moving_average(data, 500)
    print(f"Calculated {len(ma500)} MA500 values.")
    
    # Perform complex mathematical operations
    print("Performing complex mathematical operations...")
    complex_results = calculate_complex_math(data)
    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()