Spaces:
Runtime error
Runtime error
| import gradio as gr | |
| from gradio.components import File as InputFile, Number as InputNumber, Checkbox as InputCheckbox | |
| from gradio.components import File as OutputFile | |
| import yfinance as yf | |
| import pandas as pd | |
| import pandas_ta as ta | |
| from scipy.signal import find_peaks | |
| import csv | |
| import os | |
| from scipy.signal import argrelextrema | |
| from collections import deque | |
| import numpy as np | |
| def getHigherLows(data: np.array, order=5, K=2): | |
| # Get lows | |
| low_idx = argrelextrema(data, np.less, order=order)[0] | |
| lows = data[low_idx] | |
| # Ensure consecutive lows are higher than previous lows | |
| extrema = [] | |
| ex_deque = deque(maxlen=K) | |
| for i, idx in enumerate(low_idx): | |
| if i == 0: | |
| ex_deque.append(idx) | |
| continue | |
| if lows[i] < lows[i-1]: | |
| ex_deque.clear() | |
| ex_deque.append(idx) | |
| if len(ex_deque) == K: | |
| extrema.append(ex_deque.copy()) | |
| return extrema | |
| def getLowerHighs(data: np.array, order=5, K=2): | |
| # Get highs | |
| high_idx = argrelextrema(data, np.greater, order=order)[0] | |
| highs = data[high_idx] | |
| # Ensure consecutive highs are lower than previous highs | |
| extrema = [] | |
| ex_deque = deque(maxlen=K) | |
| for i, idx in enumerate(high_idx): | |
| if i == 0: | |
| ex_deque.append(idx) | |
| continue | |
| if highs[i] > highs[i-1]: | |
| ex_deque.clear() | |
| ex_deque.append(idx) | |
| if len(ex_deque) == K: | |
| extrema.append(ex_deque.copy()) | |
| return extrema | |
| def getHigherHighs(data: np.array, order=5, K=2): | |
| # Get highs | |
| high_idx = argrelextrema(data, np.greater, order=5)[0] | |
| highs = data[high_idx] | |
| # Ensure consecutive highs are higher than previous highs | |
| extrema = [] | |
| ex_deque = deque(maxlen=K) | |
| for i, idx in enumerate(high_idx): | |
| if i == 0: | |
| ex_deque.append(idx) | |
| continue | |
| if highs[i] < highs[i-1]: | |
| ex_deque.clear() | |
| ex_deque.append(idx) | |
| if len(ex_deque) == K: | |
| extrema.append(ex_deque.copy()) | |
| return extrema | |
| def getLowerLows(data: np.array, order=5, K=2): | |
| # Get lows | |
| low_idx = argrelextrema(data, np.less, order=order)[0] | |
| lows = data[low_idx] | |
| # Ensure consecutive lows are lower than previous lows | |
| extrema = [] | |
| ex_deque = deque(maxlen=K) | |
| for i, idx in enumerate(low_idx): | |
| if i == 0: | |
| ex_deque.append(idx) | |
| continue | |
| if lows[i] > lows[i-1]: | |
| ex_deque.clear() | |
| ex_deque.append(idx) | |
| if len(ex_deque) == K: | |
| extrema.append(ex_deque.copy()) | |
| return extrema | |
| def check_trend(hist): | |
| close = hist['Close'].values | |
| order = 5 | |
| K = 2 | |
| hh = getHigherHighs(close, order, K) | |
| hl = getHigherLows(close, order, K) | |
| ll = getLowerLows(close, order, K) | |
| lh = getLowerHighs(close, order, K) | |
| # Get the most recent top and bottom labels | |
| top_labels = hh if len(hh) > 0 and hh[-1][-1] > (lh[-1][-1] if len(lh) > 0 else 0) else lh | |
| bottom_labels = hl if len(hl) > 0 and hl[-1][-1] > (ll[-1][-1] if len(ll) > 0 else 0) else ll | |
| # Check if the most recent top and bottom labels form a pair | |
| if top_labels == hh and bottom_labels == hl: | |
| return True, 'bullish' | |
| elif top_labels == lh and bottom_labels == ll: | |
| return True, 'bearish' | |
| else: | |
| return False, 'uncertain' | |
| def process_csv(csv_file, lookback, bullish_stoch_value, bearish_stoch_value, check_bullish_swing, check_bearish_swing): | |
| lookback = int(lookback) | |
| all_tickers = [] | |
| with open(csv_file.name, 'r') as file: | |
| reader = csv.reader(file) | |
| next(reader) # Skip the header | |
| for row in reader: | |
| all_tickers.append(row[0]) # Append the value in the first column | |
| ttl_tickers = len(all_tickers) | |
| bullish_tickers = [] | |
| bearish_tickers = [] | |
| for ticker in all_tickers: | |
| print(f"Processing {ticker} ({all_tickers.index(ticker)+1}/{ttl_tickers})") | |
| try: | |
| data = yf.Ticker(ticker) | |
| hist = data.history(period="1y", actions=False) | |
| if not hist.empty: | |
| hist.ta.ema(close='Close', length=20, append=True) | |
| hist.ta.ema(close='Close', length=50, append=True) | |
| hist.ta.ema(close='Close', length=100, append=True) | |
| hist.ta.sma(close='Close', length=150, append=True) | |
| stoch = hist.ta.stoch(high='High', low='Low', close='Close') | |
| trend_exists, trend_type = check_trend(hist) | |
| if all(hist['EMA_20'][-lookback:] > hist['EMA_50'][-lookback:]) and all(hist['EMA_50'][-lookback:] > hist['EMA_100'][-lookback:]) and all(hist['EMA_100'][-lookback:] > hist['SMA_150'][-lookback:]) and stoch['STOCHk_14_3_3'][-1] <= bullish_stoch_value and trend_exists and trend_type == 'bullish': | |
| bullish_tickers.append([ticker, hist['Close'][-1], hist['Volume'][-1], 'Bullish']) | |
| elif all(hist['EMA_20'][-lookback:] < hist['EMA_50'][-lookback:]) and all(hist['EMA_50'][-lookback:] < hist['EMA_100'][-lookback:]) and all(hist['EMA_100'][-lookback:] < hist['SMA_150'][-lookback:]) and stoch['STOCHk_14_3_3'][-1] >= bearish_stoch_value and trend_exists and trend_type == 'bearish': | |
| bearish_tickers.append([ticker, hist['Close'][-1], hist['Volume'][-1], 'Bearish']) | |
| except Exception as e: | |
| print(f"An error occurred with ticker {ticker}: {e}") | |
| df_bullish = pd.DataFrame(bullish_tickers, columns=['Ticker', 'Close', 'Volume', 'Trend']) | |
| df_bearish = pd.DataFrame(bearish_tickers, columns=['Ticker', 'Close', 'Volume', 'Trend']) | |
| base_filename = os.path.splitext(os.path.basename(csv_file.name))[0] | |
| output_bullish_xlsx = f'{base_filename}-bullish.xlsx' | |
| output_bearish_xlsx = f'{base_filename}-bearish.xlsx' | |
| df_bullish.to_excel(output_bullish_xlsx, index=False) | |
| df_bearish.to_excel(output_bearish_xlsx, index=False) | |
| output_bullish_txt = f'{base_filename}-bullish.txt' | |
| output_bearish_txt = f'{base_filename}-bearish.txt' | |
| with open(output_bullish_txt, 'w') as f: | |
| for ticker in df_bullish['Ticker']: | |
| f.write(f"{ticker}\n") | |
| with open(output_bearish_txt, 'w') as f: | |
| for ticker in df_bearish['Ticker']: | |
| f.write(f"{ticker}\n") | |
| return output_bullish_xlsx, output_bearish_xlsx, output_bullish_txt, output_bearish_txt | |
| iface = gr.Interface( | |
| fn=process_csv, | |
| inputs=[ | |
| InputFile(label="Upload CSV"), | |
| InputNumber(label="EMA Loockback period (days, min 5 and max 60)", value=20), | |
| InputNumber(label="Bullish Stochastic Value (equal or below)", value=30), | |
| InputNumber(label="Bearish Stochastic Value (equal or above)", value=70), | |
| InputCheckbox(label="Check Bullish Swing (HH + HL) (slower, in beta)"), | |
| InputCheckbox(label="Check Bearish Swing (LH + LL) (slower, in beta)") | |
| ], | |
| outputs=[ | |
| OutputFile(label="Download Bullish XLSX"), | |
| OutputFile(label="Download Bearish XLSX"), | |
| OutputFile(label="Download Bullish TXT"), | |
| OutputFile(label="Download Bearish TXT") | |
| ], | |
| title="Stock Analysis", | |
| description="""Upload a CSV file with a column named 'Ticker' (from TradingView or other) and | |
| get a filtered shortlist of bullish and bearish stocks in return. | |
| The tool will find 'stacked' EMAs + SMAs and check for Stochastics above or below the set values.""" | |
| ) | |
| iface.launch() | |