Spaces:
Sleeping
Sleeping
| import streamlit as st | |
| import yfinance as yf | |
| import pandas as pd | |
| import numpy as np | |
| from plotly.subplots import make_subplots | |
| import plotly.graph_objects as go | |
| # Set Streamlit page configuration (wide layout) | |
| st.set_page_config( | |
| page_title="High Frequency Price Reversals", | |
| layout="wide" | |
| ) | |
| # App title | |
| st.title("Price Reversals") | |
| # Detailed purpose description | |
| st.write("This tool identifies and analyzes **high-frequency price reversals** for a given ticker or crypto pair. It calculates the number of reversals per day, their magnitude, and the time between each reversal. The analysis runs across three live intraday intervals: 60-minute, 5-minute, and 1-minute bars. Reversals help distinguish routine market noise from shifts driven by news or fundamental changes.") | |
| # Brief purpose description | |
| with st.expander("Methodology", expanded=False): | |
| st.markdown(r""" | |
| ##### What is a Price Reversal? | |
| A **price reversal** is defined as a change in direction between consecutive candles. | |
| For example, if a bullish candle (Close > Open) is followed by a bearish candle (Close < Open), that marks a reversal. The reverse scenario is treated the same way. | |
| ##### Reversal Detection Logic | |
| 1. Label each candle as bullish or bearish: | |
| $$ | |
| \text{IsBull} = \text{Close} > \text{Open} | |
| $$ | |
| 2. A reversal occurs when the current candle has a different direction than the previous one: | |
| $$ | |
| \text{Reversal} = \text{IsBull}_t \neq \text{IsBull}_{t-1} | |
| $$ | |
| ##### Daily Reversal Counts | |
| For each date: | |
| - Count the number of reversals. | |
| - Count the number of non-reversals. | |
| - Calculate the daily reversal percentage: | |
| $$ | |
| \text{ReversalPct} = \frac{\text{ReversalCount}}{\text{ReversalCount} + \text{NonReversalCount}} | |
| $$ | |
| ##### Reversal Magnitude | |
| For each reversal, we measure how much the price changed during that candle. In other words, we calculate the absolute difference between the closing and opening prices: | |
| $$ | |
| \text{ReversalMagnitude} = |\text{Close} - \text{Open}| | |
| $$ | |
| Each day, we summarize these values by computing the smallest, median, and largest price changes. This helps us see the typical size of a reversal. | |
| ##### Time Between Reversals | |
| We also measure the duration (in minutes) between one reversal and the next. For every day, we calculate the shortest, median, and longest time gaps between reversals. This measure how long the market goes without a reversal. | |
| """, unsafe_allow_html=True) | |
| # Sidebar user inputs | |
| st.sidebar.header("User Inputs") | |
| ticker_input = st.sidebar.text_input( | |
| label="Ticker", | |
| value="CVNA", | |
| help="Enter a valid ticker symbol or cryptopair (e.g. 'CVNA', 'BTC-USD', etc.)" | |
| ) | |
| run_button = st.sidebar.button( | |
| label="Run Analysis", | |
| help="Click to generate plots for the selected ticker." | |
| ) | |
| # Execute only when "Run Analysis" is clicked | |
| if run_button: | |
| if not ticker_input.strip(): | |
| st.error("Please enter a valid ticker.") | |
| else: | |
| with st.spinner("Running analysis..."): | |
| try: | |
| # Define ticker and interval settings | |
| ticker = ticker_input.strip().upper() | |
| period_mapping = { | |
| "60m": "720d", | |
| "5m": "60d", | |
| "1m": "8d" | |
| } | |
| intervals = ["60m", "5m", "1m"] | |
| # Loop over each interval | |
| for freq in intervals: | |
| st.markdown(f"## {freq} Interval Analysis - {ticker}") | |
| if freq == "60m": | |
| st.write("Analysis using 60-minute data. This interval provides a broader view of market trends.") | |
| elif freq == "5m": | |
| st.write("Analysis using 5-minute data. This interval provides a balance between overall trends and finer details.") | |
| elif freq == "1m": | |
| st.write("Analysis using 1-minute data. This interval reveals fine details in price reversals.") | |
| period = period_mapping[freq] | |
| df = yf.download(ticker, period=period, interval=freq) | |
| # Flatten multi-level columns if needed | |
| if isinstance(df.columns, pd.MultiIndex): | |
| df.columns = df.columns.get_level_values(0) | |
| df.dropna(inplace=True) | |
| df.index.name = None | |
| # Compute reversal indicator | |
| df['IsBull'] = df['Close'] > df['Open'] | |
| df['Reversal'] = df['IsBull'] != df['IsBull'].shift(1) | |
| df['Reversal'] = df['Reversal'].fillna(False) | |
| # Prepare daily counts | |
| df['Date'] = df.index.date | |
| daily_counts = df.groupby('Date')['Reversal'].agg( | |
| ReversalCount=lambda x: x.astype(int).sum(), | |
| NonReversalCount=lambda x: (~x).astype(int).sum() | |
| ) | |
| daily_counts['ReversalPct'] = daily_counts['ReversalCount'] / ( | |
| daily_counts['ReversalCount'] + daily_counts['NonReversalCount'] | |
| ) | |
| dates = pd.to_datetime(daily_counts.index) | |
| # Reversal magnitude analysis | |
| df['ReversalMagnitude'] = np.abs(df['Close'] - df['Open']) | |
| rev_mags = df[df['Reversal']].copy() | |
| rev_mags['Date'] = rev_mags.index.date | |
| mag_stats = rev_mags.groupby('Date')['ReversalMagnitude'].agg(['min', 'median', 'max']) | |
| # Time between reversals analysis | |
| reversal_times = df.index[df['Reversal']].to_series() | |
| reversal_time_diffs = reversal_times.diff().dropna().dt.total_seconds() / 60 | |
| reversal_time_df = pd.DataFrame({ | |
| 'TimeBetween': reversal_time_diffs, | |
| 'Date': reversal_times.iloc[1:].dt.date | |
| }) | |
| time_stats = reversal_time_df.groupby('Date')['TimeBetween'].agg(['min', 'median', 'max']) | |
| # Create a figure with 4 rows (subplots) | |
| fig = make_subplots( | |
| rows=4, cols=1, | |
| shared_xaxes=True, | |
| subplot_titles=( | |
| f"{ticker} Close Price with Reversal Points", | |
| "Daily Reversal Counts & Reversal %", | |
| "Daily Reversal Magnitude Summary", | |
| "Daily Time Between Reversals (Minutes)" | |
| ), | |
| specs=[ | |
| [{}], | |
| [{"secondary_y": True}], | |
| [{}], | |
| [{}] | |
| ] | |
| ) | |
| # Subplot 1: Price with reversal markers | |
| fig.add_trace( | |
| go.Scatter( | |
| x=df.index, | |
| y=df['Close'], | |
| mode='lines', | |
| name='Close Price', | |
| line=dict(color='blue') | |
| ), | |
| row=1, col=1 | |
| ) | |
| reversal_points = df[df['Reversal']] | |
| fig.add_trace( | |
| go.Scatter( | |
| x=reversal_points.index, | |
| y=reversal_points['Close'], | |
| mode='markers', | |
| name='Reversal', | |
| marker=dict(color='red', size=6, opacity=0.5) | |
| ), | |
| row=1, col=1 | |
| ) | |
| non_reversal_points = df[~df['Reversal']] | |
| fig.add_trace( | |
| go.Scatter( | |
| x=non_reversal_points.index, | |
| y=non_reversal_points['Close'], | |
| mode='markers', | |
| name='Not Reversal', | |
| marker=dict(color='green', size=4, opacity=0.5) | |
| ), | |
| row=1, col=1 | |
| ) | |
| # Subplot 2: Daily reversal counts (stacked bar) and reversal percentage line | |
| fig.add_trace( | |
| go.Bar( | |
| x=dates, | |
| y=daily_counts['ReversalCount'], | |
| name='Reversal', | |
| marker_color='red' | |
| ), | |
| row=2, col=1, secondary_y=False | |
| ) | |
| fig.add_trace( | |
| go.Bar( | |
| x=dates, | |
| y=daily_counts['NonReversalCount'], | |
| name='Not Reversal', | |
| marker_color='green', | |
| opacity=0.6 | |
| ), | |
| row=2, col=1, secondary_y=False | |
| ) | |
| fig.add_trace( | |
| go.Scatter( | |
| x=dates, | |
| y=daily_counts['ReversalPct'], | |
| mode='lines', | |
| name='Reversal %', | |
| line=dict(color='white') | |
| ), | |
| row=2, col=1, secondary_y=True | |
| ) | |
| # Subplot 3: Reversal magnitude stats | |
| fig.add_trace( | |
| go.Scatter( | |
| x=mag_stats.index, | |
| y=mag_stats['median'], | |
| mode='lines', | |
| name='Median', | |
| line=dict(color='orange') | |
| ), | |
| row=3, col=1 | |
| ) | |
| fig.add_trace( | |
| go.Scatter( | |
| x=mag_stats.index, | |
| y=mag_stats['min'], | |
| mode='lines', | |
| name='Min', | |
| line=dict(dash='dash', color='gray') | |
| ), | |
| row=3, col=1 | |
| ) | |
| fig.add_trace( | |
| go.Scatter( | |
| x=mag_stats.index, | |
| y=mag_stats['max'], | |
| mode='lines', | |
| name='Max', | |
| line=dict(dash='dash', color='white') | |
| ), | |
| row=3, col=1 | |
| ) | |
| # Subplot 4: Time between reversals stats | |
| fig.add_trace( | |
| go.Scatter( | |
| x=time_stats.index, | |
| y=time_stats['median'], | |
| mode='lines', | |
| name='Median', | |
| line=dict(color='purple') | |
| ), | |
| row=4, col=1 | |
| ) | |
| fig.add_trace( | |
| go.Scatter( | |
| x=time_stats.index, | |
| y=time_stats['min'], | |
| mode='lines', | |
| name='Min', | |
| line=dict(dash='dash', color='gray') | |
| ), | |
| row=4, col=1 | |
| ) | |
| fig.add_trace( | |
| go.Scatter( | |
| x=time_stats.index, | |
| y=time_stats['max'], | |
| mode='lines', | |
| name='Max', | |
| line=dict(dash='dash', color='white') | |
| ), | |
| row=4, col=1 | |
| ) | |
| # Update layout with dark theme, white titles, and gridlines with alpha 0.2 | |
| fig.update_layout( | |
| height=1800, | |
| width=1700, | |
| title_text=f"{ticker} Analysis ({freq} Data)", | |
| title_font_color='white', | |
| barmode='stack', | |
| template='plotly_dark', | |
| paper_bgcolor='#0e1117', | |
| plot_bgcolor='#0e1117', | |
| legend=dict(font=dict(color='white')) | |
| ) | |
| # Update all axes to set white titles and gridlines with alpha 0.2 | |
| fig.update_xaxes( | |
| title_font_color='white', | |
| gridcolor='rgba(255, 255, 255, 0.2)' | |
| ) | |
| fig.update_yaxes( | |
| title_font_color='white', | |
| gridcolor='rgba(255, 255, 255, 0.2)' | |
| ) | |
| fig.for_each_annotation(lambda a: a.update(font=dict(color='white'))) | |
| # Display the figure in Streamlit | |
| st.plotly_chart(fig, use_container_width=True) | |
| except Exception: | |
| st.error("An error occurred. Please check your input and try again.") | |
| # Hide default Streamlit style | |
| st.markdown( | |
| """ | |
| <style> | |
| #MainMenu {visibility: hidden;} | |
| footer {visibility: hidden;} | |
| </style> | |
| """, | |
| unsafe_allow_html=True | |
| ) |