Looking up consolidation on weekly candles using ATR

weekly.py

import pandas as pd
from datetime import timedelta
import psycopg2
import numpy as np
from sqlalchemy import create_engine
import holidays
import datetime

# Database connection parameters
db_params = {
    'dbname': 'postgres',
    'user': 'postgres',
    'password': 'mysecretpassword',
    'host': 'localhost',
    'port': '5432'
}

# Connect to the PostgreSQL database
conn = psycopg2.connect(**db_params)

db_user = 'postgres'
db_password = 'mysecretpassword'
db_host = 'localhost'
db_port = '5432'
db_name = 'postgres'
table_name = 'financial_data'

# Create a connection to the PostgreSQL database
engine = create_engine(f'postgresql+psycopg2://{db_user}:{db_password}@{db_host}:{db_port}/{db_name}')

# Query to fetch financial data
query = """
SELECT ticker, volume, open, close, high, low, window_start, transactions
FROM financial_data
"""

# Load data into a pandas DataFrame
df = pd.read_sql(query, engine)

# Close the database connection
conn.close()

# Convert unix timestamps to datetime
df['window_start'] = pd.to_datetime(df['window_start'], unit='ns')

# Ensure data is sorted by ticker and window_start
df.sort_values(by=['ticker', 'window_start'], inplace=True)

# Function to get the start of the week (Monday)
def get_week_start(date):
    return date - timedelta(days=date.weekday())

# Add a column for the week start
df['week_start'] = df['window_start'].apply(get_week_start)

# Group data by ticker and week_start, aggregating required fields
weekly_df = df.groupby(['ticker', 'week_start']).agg({
    'volume': 'sum',
    'open': 'first',
    'close': 'last',
    'high': 'max',
    'low': 'min',
    'transactions': 'sum'
}).reset_index()


def is_weekend_or_holiday(date, country='US'):
    bank_holidays = holidays.country_holidays(country)
    return date.weekday() >= 5 or date in bank_holidays

starting_date = datetime.date(2024, 5, 1)

def non_holiday_dates(starting_date):
    today = datetime.date.today()
    date_list = [starting_date + datetime.timedelta(days=x) for x in range((today - starting_date).days + 1)]
    non_holiday_list = [date for date in date_list if not is_weekend_or_holiday(date)]

    return non_holiday_list


# Input parameters
range_width_percent_max = 25 / 100
range_length = 8
atr_length = 14
atr_multiplier = 1.5


def calculate_atr(df, length):
    high_low = df['high'] - df['low']
    high_close = np.abs(df['high'] - df['close'].shift())
    low_close = np.abs(df['low'] - df['close'].shift())
    true_range = np.maximum(np.maximum(high_low, high_close), low_close)
    atr = true_range.rolling(window=length, min_periods=1).mean()
    return atr


def check_consolidation(df, range_length, atr_length, atr_multiplier, range_width_percent_max):
    df['highest_high'] = df['high'].rolling(window=range_length).max()
    df['lowest_low'] = df['low'].rolling(window=range_length).min()
    df['atr'] = calculate_atr(df, atr_length)
    df['atr_range_width'] = df['atr'] * atr_multiplier
    df['max_range_width'] = df['close'].rolling(window=range_length).mean() * range_width_percent_max

    df['is_consolidating'] = (
            (df['highest_high'] - df['lowest_low'] <= df['atr_range_width']) &
            (df['highest_high'] - df['lowest_low'] <= df['max_range_width'])
    )

    return df


# Apply the consolidation check for each ticker
consolidating_tickers = []

for ticker in weekly_df['ticker'].unique():
    ticker_df = weekly_df[weekly_df['ticker'] == ticker].copy()
    ticker_df = check_consolidation(ticker_df, range_length, atr_length, atr_multiplier, range_width_percent_max)

    if ticker_df['is_consolidating'].iloc[-1]:
        # consolidating_tickers.append(ticker)
        print(f'Ticker is consolidating: {ticker}')
        print(f'{weekly_df["week_start"].iloc[-1]}')

'''
# Print the tickers that are consolidating
for ticker in consolidating_tickers:
    print(f'Ticker is consolidating: {ticker}')
'''