binance/apitest/trend_detect.py

import numpy as np

import csv

def read_crypto_csv(filename):
    """
    Converts a CSV file of cryptocurrency prices into a dictionary format.
    
    Args:
        filename (str): Path to the CSV file
        
    Returns:
        dict: {coin_symbol: list_of_prices}
    """
    crypto_data = {}
    
    with open(filename, 'r') as f:
        reader = csv.reader(f)
        headers = next(reader)  # Read header row
        
        # Verify CSV format (first column should be 'Symbol')
        if headers[0] != 'Symbol':
            raise ValueError("CSV format incorrect. First column should be 'Symbol'")
            
        # Process each row
        for row in reader:
            symbol = row[0]
            prices = [float(price) for price in row[1:]]  # Convert prices to floats
            
            crypto_data[symbol] = prices
            
    return crypto_data



def detect_trend(data, window_size=10, slope_threshold=0.1):
    """
    Detects upward/downward trends for multiple coins using moving average slopes.
    
    Args:
        data (dict): Dictionary of {coin: list_of_prices}
        window_size (int): Days to analyze for initial trend
        slope_threshold (float): Slope magnitude threshold for trend classification
    
    Returns:
        dict: {coin: {'trend': 'upward'/'downward'/'flat', 'slope': float}}
    """
    results = {}
    
    for coin, prices in data.items():
        if len(prices) < window_size:
            raise ValueError(f"Not enough data for {coin}. Needs at least {window_size} days")
            
        # Calculate moving average
        window_prices = prices[:window_size]
        x = np.arange(window_size)
        y = np.array(window_prices)
        
        # Calculate slope using linear regression
        numerator = window_size * np.sum(x*y) - np.sum(x) * np.sum(y)
        denominator = window_size * np.sum(x**2) - (np.sum(x))**2
        slope = numerator / denominator if denominator != 0 else 0
        
        # Classify trend
        if slope > slope_threshold:
            trend = 'upward'
        elif slope < -slope_threshold:
            trend = 'downward'
        else:
            trend = 'flat'
            
        results[coin] = {'trend': trend, 'slope': slope}
        
    return results

def validate_trend(data, initial_trends, window_size=10, validation_period=5, slope_threshold=0.1):
    """
    Validates trends by analyzing subsequent price movements.
    
    Args:
        data (dict): Dictionary of {coin: list_of_prices}
        initial_trends (dict): Results from detect_trend()
        window_size (int): Initial detection window size
        validation_period (int): Days after window_size to analyze
        slope_threshold (float): Slope threshold for validation
    
    Returns:
        dict: {coin: {'initial_trend': str, 'validation_trend': str, 'result': str}}
    """
    validation_results = {}
    
    for coin, prices in data.items():
        if len(prices) < window_size + validation_period:
            raise ValueError(f"Not enough data for validation for {coin}")
            
        # Get validation window prices (days 11-16 for window_size=10, validation_period=5)
        validation_prices = prices[window_size:window_size+validation_period]
        
        # Calculate validation trend
        x_val = np.arange(validation_period)
        y_val = np.array(validation_prices)
        
        numerator_val = validation_period * np.sum(x_val*y_val) - np.sum(x_val) * np.sum(y_val)
        denominator_val = validation_period * np.sum(x_val**2) - (np.sum(x_val))**2
        slope_val = numerator_val / denominator_val if denominator_val != 0 else 0
        
        # Classify validation trend
        if slope_val > slope_threshold:
            val_trend = 'upward'
        elif slope_val < -slope_threshold:
            val_trend = 'downward'
        else:
            val_trend = 'flat'
            
        # Determine validation result
        initial = initial_trends[coin]['trend']
        
        if initial == 'flat':
            if val_trend != 'flat':
                result = f'New {val_trend} trend emerged'
            else:
                result = 'Continued flat'
        else:
            if val_trend == initial:
                result = 'Trend continued'
            elif val_trend == 'flat':
                result = 'Trend weakened to flat'
            else:
                result = 'Trend reversed'
                
        validation_results[coin] = {
            'initial_trend': initial,
            'validation_trend': val_trend,
            'result': result,
            'validation_slope': slope_val
        }
        
    return validation_results

# Usage example
if __name__ == '__main__':
    # Convert CSV to crypto_data structure
    crypto_data = read_crypto_csv('crypto_prices.csv')

    # Detect initial trends
    initial_results = detect_trend(crypto_data, window_size=10)

    # Validate trends using next 5 days
    validation_results = validate_trend(crypto_data, initial_results, window_size=10, validation_period=5)

    # Print formatted results
    print("Coin\t\tInitial\t\tValidation\tResult")
    print("-----------------------------------------------------------")
    for coin, result in validation_results.items():
        print(f"{coin}\t{result['initial_trend']:8}\t{result['validation_trend']:8}\t{result['result']}")

    # # Example: Print first 3 coins and their first 5 prices
    # for i, (coin, prices) in enumerate(crypto_data.items()):
    #     if i >= 3:
    #         break
    #     print(f"{coin}: {prices[:5]}...")