Created
May 3, 2026 13:49
-
-
Save dmzoneill/64a18fe75765d2a7b672361b0c871150 to your computer and use it in GitHub Desktop.
Crude oil price monitor with technical analysis indicators in Python
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| """Crude oil price monitor with technical analysis indicators.""" | |
| import argparse | |
| import math | |
| import shutil | |
| import subprocess | |
| import sys | |
| import time | |
| from dataclasses import dataclass | |
| from datetime import datetime | |
| from enum import Enum | |
| import pandas as pd | |
| import ta | |
| import yfinance as yf | |
| TICKER = "CL=F" | |
| INTERVAL_TO_PERIOD = { | |
| "1m": "5d", | |
| "2m": "5d", | |
| "5m": "1mo", | |
| "15m": "1mo", | |
| "30m": "1mo", | |
| "1h": "3mo", | |
| "1d": "6mo", | |
| } | |
| INTERVAL_TO_SECONDS = { | |
| "1m": 60, | |
| "2m": 120, | |
| "5m": 300, | |
| "15m": 900, | |
| "30m": 1800, | |
| "1h": 3600, | |
| "1d": 86400, | |
| } | |
| INDICATOR_WEIGHTS = { | |
| "SMA (20/50)": 1.5, | |
| "EMA (12/26)": 1.5, | |
| "RSI (14)": 1.0, | |
| "MACD": 2.0, | |
| "Bollinger Bands": 1.0, | |
| "Stochastic": 1.0, | |
| "ATR (14)": 0.0, | |
| "OBV": 1.0, | |
| "CCI (20)": 1.0, | |
| "Williams %R": 1.0, | |
| } | |
| GREEN = "\033[92m" | |
| RED = "\033[91m" | |
| YELLOW = "\033[93m" | |
| CYAN = "\033[96m" | |
| BOLD = "\033[1m" | |
| DIM = "\033[2m" | |
| RESET = "\033[0m" | |
| class Signal(Enum): | |
| BUY = "BUY" | |
| SELL = "SELL" | |
| NEUTRAL = "NEUTRAL" | |
| @dataclass | |
| class IndicatorResult: | |
| name: str | |
| value: str | |
| signal: Signal | |
| def safe_val(series, offset=-1): | |
| try: | |
| v = series.iloc[offset] | |
| if pd.isna(v) or (isinstance(v, float) and math.isnan(v)): | |
| return None | |
| return float(v) | |
| except (IndexError, TypeError): | |
| return None | |
| def fetch_data(interval: str) -> pd.DataFrame: | |
| period = INTERVAL_TO_PERIOD[interval] | |
| df = yf.download(TICKER, period=period, interval=interval, progress=False, multi_level_index=False) | |
| if df.empty: | |
| raise RuntimeError("No data returned from Yahoo Finance") | |
| return df | |
| def compute_sma(df: pd.DataFrame) -> IndicatorResult: | |
| sma20 = ta.trend.sma_indicator(df["Close"], window=20) | |
| sma50 = ta.trend.sma_indicator(df["Close"], window=50) | |
| v20, v50 = safe_val(sma20), safe_val(sma50) | |
| price = safe_val(df["Close"]) | |
| if v20 is None or v50 is None or price is None: | |
| return IndicatorResult("SMA (20/50)", "N/A", Signal.NEUTRAL) | |
| value = f"20: {v20:.2f} | 50: {v50:.2f}" | |
| if price > v20 and v20 > v50: | |
| signal = Signal.BUY | |
| elif price < v20 and v20 < v50: | |
| signal = Signal.SELL | |
| else: | |
| signal = Signal.NEUTRAL | |
| return IndicatorResult("SMA (20/50)", value, signal) | |
| def compute_ema(df: pd.DataFrame) -> IndicatorResult: | |
| ema12 = ta.trend.ema_indicator(df["Close"], window=12) | |
| ema26 = ta.trend.ema_indicator(df["Close"], window=26) | |
| v12, v26 = safe_val(ema12), safe_val(ema26) | |
| if v12 is None or v26 is None: | |
| return IndicatorResult("EMA (12/26)", "N/A", Signal.NEUTRAL) | |
| value = f"12: {v12:.2f} | 26: {v26:.2f}" | |
| if v12 > v26: | |
| signal = Signal.BUY | |
| elif v12 < v26: | |
| signal = Signal.SELL | |
| else: | |
| signal = Signal.NEUTRAL | |
| return IndicatorResult("EMA (12/26)", value, signal) | |
| def compute_rsi(df: pd.DataFrame) -> IndicatorResult: | |
| rsi = ta.momentum.rsi(df["Close"], window=14) | |
| v = safe_val(rsi) | |
| if v is None: | |
| return IndicatorResult("RSI (14)", "N/A", Signal.NEUTRAL) | |
| value = f"{v:.1f}" | |
| if v < 30: | |
| signal = Signal.BUY | |
| elif v > 70: | |
| signal = Signal.SELL | |
| else: | |
| signal = Signal.NEUTRAL | |
| return IndicatorResult("RSI (14)", value, signal) | |
| def compute_macd(df: pd.DataFrame) -> IndicatorResult: | |
| macd_line = ta.trend.macd(df["Close"]) | |
| macd_signal = ta.trend.macd_signal(df["Close"]) | |
| macd_hist = ta.trend.macd_diff(df["Close"]) | |
| vl, vs, vh = safe_val(macd_line), safe_val(macd_signal), safe_val(macd_hist) | |
| if vl is None or vs is None or vh is None: | |
| return IndicatorResult("MACD", "N/A", Signal.NEUTRAL) | |
| value = f"{vl:.2f} | Sig: {vs:.2f} | H: {vh:.2f}" | |
| if vl > vs and vh > 0: | |
| signal = Signal.BUY | |
| elif vl < vs and vh < 0: | |
| signal = Signal.SELL | |
| else: | |
| signal = Signal.NEUTRAL | |
| return IndicatorResult("MACD", value, signal) | |
| def compute_bollinger(df: pd.DataFrame) -> IndicatorResult: | |
| bb = ta.volatility.BollingerBands(df["Close"], window=20, window_dev=2) | |
| upper = safe_val(bb.bollinger_hband()) | |
| mid = safe_val(bb.bollinger_mavg()) | |
| lower = safe_val(bb.bollinger_lband()) | |
| price = safe_val(df["Close"]) | |
| if upper is None or lower is None or price is None or mid is None: | |
| return IndicatorResult("Bollinger Bands", "N/A", Signal.NEUTRAL) | |
| value = f"U: {upper:.2f} M: {mid:.2f} L: {lower:.2f}" | |
| if price <= lower: | |
| signal = Signal.BUY | |
| elif price >= upper: | |
| signal = Signal.SELL | |
| else: | |
| signal = Signal.NEUTRAL | |
| return IndicatorResult("Bollinger Bands", value, signal) | |
| def compute_stochastic(df: pd.DataFrame) -> IndicatorResult: | |
| stoch = ta.momentum.StochasticOscillator(df["High"], df["Low"], df["Close"]) | |
| k = safe_val(stoch.stoch()) | |
| d = safe_val(stoch.stoch_signal()) | |
| if k is None or d is None: | |
| return IndicatorResult("Stochastic", "N/A", Signal.NEUTRAL) | |
| value = f"%K: {k:.1f} | %D: {d:.1f}" | |
| if k < 20: | |
| signal = Signal.BUY | |
| elif k > 80: | |
| signal = Signal.SELL | |
| else: | |
| signal = Signal.NEUTRAL | |
| return IndicatorResult("Stochastic", value, signal) | |
| def compute_atr(df: pd.DataFrame) -> IndicatorResult: | |
| atr = ta.volatility.AverageTrueRange(df["High"], df["Low"], df["Close"], window=14) | |
| v = safe_val(atr.average_true_range()) | |
| if v is None: | |
| return IndicatorResult("ATR (14)", "N/A", Signal.NEUTRAL) | |
| return IndicatorResult("ATR (14)", f"{v:.2f}", Signal.NEUTRAL) | |
| def compute_obv(df: pd.DataFrame) -> IndicatorResult: | |
| obv = ta.volume.on_balance_volume(df["Close"], df["Volume"]) | |
| if len(obv) < 20: | |
| return IndicatorResult("OBV", "N/A", Signal.NEUTRAL) | |
| obv_sma5 = obv.rolling(window=5).mean() | |
| obv_sma20 = obv.rolling(window=20).mean() | |
| s5, s20 = safe_val(obv_sma5), safe_val(obv_sma20) | |
| v = safe_val(obv) | |
| if v is None or s5 is None or s20 is None: | |
| return IndicatorResult("OBV", "N/A", Signal.NEUTRAL) | |
| if abs(v) >= 1_000_000: | |
| display = f"{v / 1_000_000:.1f}M" | |
| elif abs(v) >= 1_000: | |
| display = f"{v / 1_000:.1f}K" | |
| else: | |
| display = f"{v:.0f}" | |
| if s5 > s20: | |
| signal = Signal.BUY | |
| elif s5 < s20: | |
| signal = Signal.SELL | |
| else: | |
| signal = Signal.NEUTRAL | |
| return IndicatorResult("OBV", display, signal) | |
| def compute_cci(df: pd.DataFrame) -> IndicatorResult: | |
| cci = ta.trend.cci(df["High"], df["Low"], df["Close"], window=20) | |
| v = safe_val(cci) | |
| if v is None: | |
| return IndicatorResult("CCI (20)", "N/A", Signal.NEUTRAL) | |
| value = f"{v:.1f}" | |
| if v < -100: | |
| signal = Signal.BUY | |
| elif v > 100: | |
| signal = Signal.SELL | |
| else: | |
| signal = Signal.NEUTRAL | |
| return IndicatorResult("CCI (20)", value, signal) | |
| def compute_williams_r(df: pd.DataFrame) -> IndicatorResult: | |
| wr = ta.momentum.williams_r(df["High"], df["Low"], df["Close"], lbp=14) | |
| v = safe_val(wr) | |
| if v is None: | |
| return IndicatorResult("Williams %R", "N/A", Signal.NEUTRAL) | |
| value = f"{v:.1f}" | |
| if v < -80: | |
| signal = Signal.BUY | |
| elif v > -20: | |
| signal = Signal.SELL | |
| else: | |
| signal = Signal.NEUTRAL | |
| return IndicatorResult("Williams %R", value, signal) | |
| def compute_all_indicators(df: pd.DataFrame) -> list[IndicatorResult]: | |
| return [ | |
| compute_sma(df), | |
| compute_ema(df), | |
| compute_rsi(df), | |
| compute_macd(df), | |
| compute_bollinger(df), | |
| compute_stochastic(df), | |
| compute_atr(df), | |
| compute_obv(df), | |
| compute_cci(df), | |
| compute_williams_r(df), | |
| ] | |
| def aggregate_signals(indicators: list[IndicatorResult]) -> tuple[str, float, str]: | |
| weighted_sum = 0.0 | |
| max_weight = 0.0 | |
| buys, sells, neutrals = 0, 0, 0 | |
| for ind in indicators: | |
| w = INDICATOR_WEIGHTS.get(ind.name, 1.0) | |
| if w == 0: | |
| continue | |
| max_weight += w | |
| if ind.signal == Signal.BUY: | |
| weighted_sum += w | |
| buys += 1 | |
| elif ind.signal == Signal.SELL: | |
| weighted_sum -= w | |
| sells += 1 | |
| else: | |
| neutrals += 1 | |
| if max_weight == 0: | |
| return "NEUTRAL", 0.0, "0B/0S/0N" | |
| confidence = abs(weighted_sum) / max_weight * 100 | |
| vote_summary = f"{buys}B/{sells}S/{neutrals}N" | |
| if confidence >= 70: | |
| label = "STRONG BUY" if weighted_sum > 0 else "STRONG SELL" | |
| elif confidence >= 40: | |
| label = "BUY" if weighted_sum > 0 else "SELL" | |
| else: | |
| label = "NEUTRAL" | |
| return label, confidence, vote_summary | |
| def send_notification(title: str, message: str) -> None: | |
| if not shutil.which("notify-send"): | |
| return | |
| try: | |
| subprocess.run( | |
| ["notify-send", "--urgency=critical", "--icon=dialog-warning", title, message], | |
| capture_output=True, | |
| timeout=5, | |
| ) | |
| except Exception: | |
| pass | |
| def check_price_alert( | |
| current: float, previous: float | None, threshold: float, recommendation: str | |
| ) -> None: | |
| if previous is None: | |
| return | |
| pct = (current - previous) / previous * 100 | |
| if abs(pct) >= threshold: | |
| direction = "UP" if pct > 0 else "DOWN" | |
| title = f"Crude Oil Alert: {direction} {abs(pct):.2f}%" | |
| message = ( | |
| f"${previous:.2f} -> ${current:.2f}\n" | |
| f"Recommendation: {recommendation}" | |
| ) | |
| send_notification(title, message) | |
| def signal_icon(s: Signal) -> str: | |
| if s == Signal.BUY: | |
| return f"{GREEN}{BOLD}▲ BUY{RESET}" | |
| elif s == Signal.SELL: | |
| return f"{RED}{BOLD}▼ SELL{RESET}" | |
| return f"{YELLOW}─ NEUTRAL{RESET}" | |
| def recommendation_color(label: str) -> str: | |
| if "BUY" in label: | |
| return f"{GREEN}{BOLD}{label}{RESET}" | |
| elif "SELL" in label: | |
| return f"{RED}{BOLD}{label}{RESET}" | |
| return f"{YELLOW}{BOLD}{label}{RESET}" | |
| def format_dashboard( | |
| price: float, | |
| prev_price: float | None, | |
| indicators: list[IndicatorResult], | |
| rec_label: str, | |
| confidence: float, | |
| vote_summary: str, | |
| interval: str, | |
| poll_seconds: int, | |
| market_open: bool, | |
| ) -> str: | |
| now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") | |
| w = 72 | |
| top = f"╔{'═' * w}╗" | |
| sep = f"╠{'═' * w}╣" | |
| bot = f"╚{'═' * w}╝" | |
| lines = [top] | |
| title = f" CRUDE OIL MONITOR ({TICKER})" | |
| market_tag = "" if market_open else f" {DIM}(Market Closed){RESET}" | |
| ts = f"{now}" | |
| title_vis = len(f" CRUDE OIL MONITOR ({TICKER})") + (len(" (Market Closed)") if not market_open else 0) | |
| right_pad = w - title_vis - len(ts) | |
| header = f"║{CYAN}{BOLD}{title}{RESET}{market_tag}{' ' * max(1, right_pad - (len(' (Market Closed)') if not market_open else 0))}{DIM}{ts}{RESET}║" | |
| lines.append(header) | |
| lines.append(sep) | |
| if prev_price is not None: | |
| change = price - prev_price | |
| pct = (change / prev_price) * 100 | |
| sign = "+" if change >= 0 else "" | |
| color = GREEN if change >= 0 else RED | |
| price_line = f" Current Price: ${price:.2f} | Change: {color}{sign}{change:.2f} ({sign}{pct:.2f}%){RESET}" | |
| price_vis = len(f" Current Price: ${price:.2f} | Change: {sign}{change:.2f} ({sign}{pct:.2f}%)") | |
| else: | |
| price_line = f" Current Price: ${price:.2f} | Change: {DIM}N/A{RESET}" | |
| price_vis = len(f" Current Price: ${price:.2f} | Change: N/A") | |
| lines.append(f"║{price_line}{' ' * max(0, w - price_vis)}║") | |
| lines.append(sep) | |
| col1 = "INDICATOR" | |
| col2 = "VALUE" | |
| col3 = "SIGNAL" | |
| header_row = f" {col1:<20}{col2:<32}{col3}" | |
| lines.append(f"║{header_row}{' ' * max(0, w - len(header_row))}║") | |
| lines.append(f"║ {'─' * (w - 4)} ║") | |
| for ind in indicators: | |
| sig = signal_icon(ind.signal) | |
| sig_plain_len = len(f"▲ {ind.signal.value}") if ind.signal != Signal.NEUTRAL else len(f"─ NEUTRAL") | |
| row_text = f" {ind.name:<20}{ind.value:<32}" | |
| row_vis_len = len(row_text) + sig_plain_len | |
| lines.append(f"║{row_text}{sig}{' ' * max(0, w - row_vis_len)}║") | |
| lines.append(sep) | |
| rec_colored = recommendation_color(rec_label) | |
| rec_line = f" RECOMMENDATION: {rec_colored} | Confidence: {confidence:.1f}% [{vote_summary}]" | |
| rec_vis_len = len(f" RECOMMENDATION: {rec_label} | Confidence: {confidence:.1f}% [{vote_summary}]") | |
| lines.append(f"║{rec_line}{' ' * max(0, w - rec_vis_len)}║") | |
| refresh_line = f" Next refresh in {poll_seconds}s ({interval} interval)" | |
| lines.append(f"║{refresh_line}{' ' * max(0, w - len(refresh_line))}║") | |
| lines.append(bot) | |
| return "\n".join(lines) | |
| def is_market_open(df: pd.DataFrame) -> bool: | |
| if df.index.empty: | |
| return False | |
| last_ts = pd.Timestamp(df.index[-1]) | |
| last_epoch = last_ts.timestamp() | |
| age_seconds = time.time() - last_epoch | |
| return age_seconds < 7200 | |
| def parse_args() -> argparse.Namespace: | |
| parser = argparse.ArgumentParser( | |
| description="Crude Oil Price Monitor with Technical Analysis", | |
| formatter_class=argparse.RawDescriptionHelpFormatter, | |
| epilog=( | |
| "Intervals: 1m, 2m, 5m, 15m, 30m, 1h, 1d\n" | |
| "Example: python3 crudeoil.py --interval 5m --alert-pct 1.5" | |
| ), | |
| ) | |
| parser.add_argument( | |
| "--interval", | |
| default="15m", | |
| choices=list(INTERVAL_TO_SECONDS.keys()), | |
| help="polling and candle interval (default: 15m)", | |
| ) | |
| parser.add_argument( | |
| "--alert-pct", | |
| type=float, | |
| default=1.0, | |
| help="percentage change threshold for desktop notifications (default: 1.0)", | |
| ) | |
| return parser.parse_args() | |
| def main() -> None: | |
| args = parse_args() | |
| interval = args.interval | |
| alert_pct = args.alert_pct | |
| poll_seconds = INTERVAL_TO_SECONDS[interval] | |
| if not shutil.which("notify-send"): | |
| print(f"{YELLOW}Warning: notify-send not found — desktop notifications disabled{RESET}") | |
| print(f"Starting Crude Oil Monitor | Interval: {interval} | Alert: {alert_pct}%") | |
| print("Fetching initial data...") | |
| previous_price = None | |
| while True: | |
| try: | |
| df = fetch_data(interval) | |
| price = float(df["Close"].iloc[-1]) | |
| market_open = is_market_open(df) | |
| indicators = compute_all_indicators(df) | |
| rec_label, confidence, vote_summary = aggregate_signals(indicators) | |
| check_price_alert(price, previous_price, alert_pct, rec_label) | |
| print("\033[2J\033[H", end="") | |
| print(format_dashboard( | |
| price, previous_price, indicators, | |
| rec_label, confidence, vote_summary, | |
| interval, poll_seconds, market_open, | |
| )) | |
| previous_price = price | |
| time.sleep(poll_seconds) | |
| except KeyboardInterrupt: | |
| print(f"\n{DIM}Exiting...{RESET}") | |
| sys.exit(0) | |
| except Exception as e: | |
| print(f"\n{RED}Error: {e}{RESET}") | |
| print(f"Retrying in {poll_seconds}s...") | |
| time.sleep(poll_seconds) | |
| if __name__ == "__main__": | |
| main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment