From 3de5c0778e7fe25f067ef475f7bc0068749ba875 Mon Sep 17 00:00:00 2001 From: Valentin Richter Date: Mon, 1 Jun 2026 03:39:52 -0400 Subject: [PATCH] create dashboard, augment data in Gantt chart --- .github/workflows/ci-test.yml | 8 +- README.md | 83 +++++++ auth.py | 9 +- data.py | 146 +++++------- main.py | 7 +- plot.py | 435 +++++++++++++++++++++++++++++----- requirements.txt | 1 + streamlit_app.py | 18 ++ test.py | 269 ++++++++++++++++++--- 9 files changed, 793 insertions(+), 183 deletions(-) create mode 100644 README.md create mode 100644 streamlit_app.py diff --git a/.github/workflows/ci-test.yml b/.github/workflows/ci-test.yml index 2f7a78b..4ccff54 100644 --- a/.github/workflows/ci-test.yml +++ b/.github/workflows/ci-test.yml @@ -9,14 +9,18 @@ on: jobs: test: runs-on: ubuntu-latest + strategy: + matrix: + python-version: ['3.9', '3.10', '3.11'] + steps: - name: Checkout uses: actions/checkout@v4 - - name: Set up Python + - name: Set up Python ${{ matrix.python-version }} uses: actions/setup-python@v4 with: - python-version: '3.x' + python-version: ${{ matrix.python-version }} - name: Install dependencies run: | diff --git a/README.md b/README.md new file mode 100644 index 0000000..3626851 --- /dev/null +++ b/README.md @@ -0,0 +1,83 @@ +# optchart + +Lightweight tooling to collect and visualize option position data from Schwab via OAuth. Displays a Gantt chart of position lifespans, intrinsic/extrinsic values, and market tracking. + +## Features + +- **OAuth Authentication**: Secure login to Schwab API with automatic token refresh +- **Position Data**: Fetch current option positions and account info +- **Gantt Visualization**: Timeline chart showing option expiration dates with opacity proportional to profitability +- **Value Analysis**: Displays intrinsic and extrinsic values in a live dashboard table +- **Local Caching**: Stores position and market data locally for offline review + +## Prerequisites + +- Python 3.8+ +- Schwab API credentials (obtain from [Schwab Developer Portal](https://developer.schwab.com)) + +## Installation + +```bash +git clone +cd optchart +python -m pip install -r requirements.txt +``` + +## Quick Start + +### 1. Generate Credentials + +Obtain your `client_id` and `client_secret` from Schwab, then save to `auth/client.json`: + +```json +{ + "client_id": "YOUR_CLIENT_ID", + "client_secret": "YOUR_CLIENT_SECRET" +} +``` + +### 2. Fetch Data and Generate Chart + +```bash +python main.py +``` + +This will: +- Authenticate with Schwab (browser login on first run) +- Fetch account positions and option prices +- Generate `data/gantt.png` showing position timelines +- Cache data locally in `data/positions.json` and `data/tracking.json` + +### 3. View the Dashboard + +```bash +streamlit run streamlit_app.py +``` + +Opens an interactive dashboard at `http://localhost:8501` showing the Gantt chart and values table. + +## Testing + +Run integrity checks: + +```bash +python test.py +``` + +Verifies Python syntax, JSON structures, .gitignore protections, and chart generation. + +## Security + +- **Local-Only Credentials**: `auth/` folder is git-ignored; credentials never leave your machine +- **Token Refresh**: Automatic refresh handles token expiry gracefully +- **No Remote Storage**: All data cached locally only + +## File Structure + +- `main.py`: Entry point; orchestrates auth, data fetch, and chart generation +- `auth.py`: OAuth flow and token management +- `data.py`: Schwab API client; fetches positions and quotes +- `plot.py`: Gantt chart generation and Streamlit dashboard +- `test.py`: Integration tests and project validation +- `auth/`: Local OAuth credentials (git-ignored) +- `data/`: Cached positions, market data, and generated chart (git-ignored) diff --git a/auth.py b/auth.py index e63faa9..71af33a 100644 --- a/auth.py +++ b/auth.py @@ -226,7 +226,6 @@ def perform_initial_handshake(token_file: str = TOKEN_FILE) -> Dict[str, Any]: print("Waiting for Schwab login…") code = get_auth_code(client_id) - print("Authorization code received:", code) raw_tokens = exchange_code_for_tokens(code, client_id, client_secret) save_tokens(raw_tokens, token_file) @@ -264,8 +263,12 @@ def refresh_access_token(tokens: Dict[str, Any], token_file: str = TOKEN_FILE) - print("Refreshing access token...") resp = requests.post(TOKEN_URL, headers=headers, data=data) - resp.raise_for_status() - + try: + resp.raise_for_status() + except requests.exceptions.HTTPError as exc: + print(f"Refresh token request failed ({resp.status_code}): {resp.text}") + raise + new_tokens = resp.json() # Preserve refresh_token if not included in response if "refresh_token" not in new_tokens and "refresh_token" in tokens: diff --git a/data.py b/data.py index 57a851b..1a5eacb 100644 --- a/data.py +++ b/data.py @@ -163,38 +163,21 @@ def _parse_account_numbers(data: Any) -> List[Dict[str, Any]]: def fetch_accounts(session: requests.Session) -> List[Dict[str, Any]]: - urls = [ - f"{TRADER_API_BASE}/accounts/accountNumbers", - f"{TRADER_API_BASE}/accounts", - f"{TRADER_API_BASE}/accounts/", - f"{TRADER_API_BASE}/accounts/list", - ] - - data = None - for url in urls: - try: - resp = session.get(url) - except Exception as exc: - print(f"Failed to request {url}: {exc}") - continue - - if resp.status_code == 404: - # try next candidate URL - print(f"Endpoint not found: {url} (404)") - continue + url = f"{TRADER_API_BASE}/accounts/accountNumbers" + try: + resp = session.get(url) + resp.raise_for_status() + except Exception as exc: + print(f"Error fetching accounts from Schwab API: {exc}") + return [] - try: - resp.raise_for_status() - except Exception as exc: - print(f"Error fetching accounts from {url}: {exc} - {getattr(resp, 'text', '')}") - continue + try: + data = resp.json() + except Exception as exc: + print(f"Failed to decode account JSON from Schwab response: {exc}") + return [] - try: - data = resp.json() - break - except Exception as exc: - print(f"Failed to decode JSON from {url}: {exc}") - continue + # accounts successfully fetched if data is None: print("No accounts data available from Schwab API; returning empty list.") @@ -426,24 +409,20 @@ def fetch_quotes(session: requests.Session, symbols: List[str]) -> Dict[str, Dic return {} symbol_list = ",".join(symbols) - urls = [ - f"{MARKETDATA_API_BASE}/quotes?symbols={symbol_list}", - f"{MARKETDATA_API_BASE}/quotes?symbols={symbol_list}", - ] - - for url in urls: - resp = session.get(url) - if resp.status_code != 200: - continue + url = f"{MARKETDATA_API_BASE}/quotes?symbols={symbol_list}" + resp = session.get(url) + if resp.status_code != 200: + print(f"Unable to fetch quotes from Schwab API: {resp.status_code}") + return {} - data = resp.json() - if isinstance(data, dict): - if "quotes" in data and isinstance(data["quotes"], list): - return {q["symbol"]: q for q in data["quotes"] if isinstance(q, dict) and isinstance(q.get("symbol"), str)} - if "data" in data and isinstance(data["data"], list): - return {q["symbol"]: q for q in data["data"] if isinstance(q, dict) and isinstance(q.get("symbol"), str)} - if all(isinstance(v, dict) for v in data.values()): - return {k: v for k, v in data.items() if isinstance(v, dict)} + data = resp.json() + if isinstance(data, dict): + if "quotes" in data and isinstance(data["quotes"], list): + return {q["symbol"]: q for q in data["quotes"] if isinstance(q, dict) and isinstance(q.get("symbol"), str)} + if "data" in data and isinstance(data["data"], list): + return {q["symbol"]: q for q in data["data"] if isinstance(q, dict) and isinstance(q.get("symbol"), str)} + if all(isinstance(v, dict) for v in data.values()): + return {k: v for k, v in data.items() if isinstance(v, dict)} print("Unable to fetch quotes for symbols:", symbol_list) return {} @@ -458,46 +437,45 @@ def fetch_earnings(session: requests.Session, symbols: List[str]) -> Dict[str, D return {} symbol_list = ",".join(symbols) - urls = [ - f"{MARKETDATA_API_BASE}/earnings?symbols={symbol_list}", - f"{MARKETDATA_API_BASE}/earnings?symbols={symbol_list}", - f"{MARKETDATA_API_BASE}/earnings", - ] + url = f"{MARKETDATA_API_BASE}/earnings?symbols={symbol_list}" + try: + resp = session.get(url) + except Exception as exc: + print(f"Unable to fetch earnings from Schwab API: {exc}") + return {} - for url in urls: - try: - resp = session.get(url) - except Exception: - continue - if resp.status_code != 200: - continue + if resp.status_code != 200: + print(f"Unable to fetch earnings from Schwab API: {resp.status_code}") + return {} - try: - data = resp.json() - except Exception as exc: - continue + try: + data = resp.json() + except Exception as exc: + print(f"Failed to decode earnings JSON from Schwab: {exc}") + return {} - # Common shapes: {"earnings": [ {"symbol": "AAPL", ...}, ... ]} - if isinstance(data, dict): - if "earnings" in data and isinstance(data["earnings"], list): - earnings = {} - for e in data["earnings"]: - if isinstance(e, dict): - symbol = e.get("symbol") - if isinstance(symbol, str): - earnings[symbol] = e - return earnings - if "data" in data and isinstance(data["data"], list): - earnings = {} - for e in data["data"]: - if isinstance(e, dict): - symbol = e.get("symbol") - if isinstance(symbol, str): - earnings[symbol] = e - return earnings - # If API returns a dict of symbol->info - if all(isinstance(v, dict) for v in data.values()): - return {k: v for k, v in data.items() if isinstance(v, dict)} + if isinstance(data, dict): + if "earnings" in data and isinstance(data["earnings"], list): + earnings = {} + for e in data["earnings"]: + if isinstance(e, dict): + symbol = e.get("symbol") + if isinstance(symbol, str): + earnings[symbol] = e + # earnings fetched + return earnings + if "data" in data and isinstance(data["data"], list): + earnings = {} + for e in data["data"]: + if isinstance(e, dict): + symbol = e.get("symbol") + if isinstance(symbol, str): + earnings[symbol] = e + # earnings fetched + return earnings + if all(isinstance(v, dict) for v in data.values()): + # earnings fetched + return {k: v for k, v in data.items() if isinstance(v, dict)} # No earnings data available from tried endpoints return {} diff --git a/main.py b/main.py index 2966de0..e952b0f 100644 --- a/main.py +++ b/main.py @@ -29,7 +29,12 @@ def main() -> None: while True: # Check if token needs refreshing if auth.is_token_expired(tokens): - tokens = auth.refresh_access_token(tokens) + try: + tokens = auth.refresh_access_token(tokens) + except Exception as exc: + print(f"Token refresh failed: {exc}") + print("Attempting full authentication flow...") + tokens = auth.perform_initial_handshake() if iteration % 5 == 0: data.run() diff --git a/plot.py b/plot.py index 3ac5295..fd81025 100644 --- a/plot.py +++ b/plot.py @@ -37,6 +37,254 @@ def import_plotting() -> tuple[Any, Any, Any]: OUTPUT_FILE = os.path.join(DATA_DIR, "gantt.png") +def _format_money(v: Optional[float]) -> str: + if v is None: + return "N/A" + try: + rounded = round(float(v), 2) + return f"${rounded:,.2f}" + except Exception: + return str(v) + +FONT_FAMILY = "DejaVu Sans, Arial, sans-serif" +POSITIVE_COLOR = "#00ff00" +NEGATIVE_COLOR = "#ff0000" + + +def _format_money_html(v: Optional[float]) -> str: + if v is None: + return "N/A" + try: + value = float(v) + except Exception: + return str(v) + + rounded = round(value, 2) + # If the value rounds to zero at two decimal places, show as $0.00 + if rounded == 0.0: + return f"${abs(rounded):,.2f}" + if rounded < 0: + return f"(${abs(rounded):,.2f})" + return f"${rounded:,.2f}" + + +def _money_color(v: Optional[float]) -> str: + if v is None: + return "white" + try: + value = float(v) + except Exception: + return "white" + rounded = round(value, 2) + # Values that round to $0.00 should be neutral/white + if rounded == 0.0: + return "white" + if value < 0: + return NEGATIVE_COLOR + return POSITIVE_COLOR + + +def _compute_option_values(option: Dict[str, Any], tracking_data: Dict[str, Any]) -> Dict[str, Optional[float]]: + underlying = option.get("underlying") + strike = option.get("strike") + put_call = option.get("put_call") + quantity = option.get("quantity") + market_value = option.get("market_value") + underlying_mark = None + + if underlying and isinstance(tracking_data, dict) and underlying in tracking_data: + try: + underlying_mark = tracking_data[underlying].get("mark") + except Exception: + underlying_mark = None + + intrinsic_per = None + if isinstance(underlying_mark, (int, float)) and isinstance(strike, (int, float)): + if isinstance(put_call, str) and put_call.upper().startswith("C"): + intrinsic_per = max(0.0, underlying_mark - float(strike)) + elif isinstance(put_call, str) and put_call.upper().startswith("P"): + intrinsic_per = max(0.0, float(strike) - underlying_mark) + + premium_per = None + if isinstance(market_value, (int, float)) and isinstance(quantity, (int, float)) and quantity != 0: + premium_per = abs(market_value) / abs(quantity) + + sign = 1 if isinstance(quantity, (int, float)) and quantity >= 0 else -1 + intrinsic_total = None + extrinsic_total = None + if intrinsic_per is not None and isinstance(quantity, (int, float)): + intrinsic_total = sign * intrinsic_per * abs(quantity) + if intrinsic_per is not None and premium_per is not None and isinstance(quantity, (int, float)): + extrinsic_total = sign * (premium_per - intrinsic_per) * abs(quantity) + + return { + "intrinsic": intrinsic_total, + "extrinsic": extrinsic_total, + } + + +def _ordered_option_groups(positions_data: Dict[str, Any], tracking_data: Dict[str, Any]) -> List[Tuple[str, List[Dict[str, Any]]]]: + grouped_options: Dict[str, List[Dict[str, Any]]] = {} + + for entry in positions_data.get("summary", []): + if not isinstance(entry, dict): + continue + + instrument = entry.get("instrument") or {} + symbol = entry.get("symbol") or instrument.get("symbol") + underlying = instrument.get("underlyingSymbol") or entry.get("underlying_symbol") or "UNKNOWN" + strike = entry.get("strike_price") or instrument.get("strikePrice") + put_call = (instrument.get("putCall") or entry.get("type") or "").upper() + init_date = entry.get("init") + expiration_date = entry.get("expiration_date") + + if not init_date or not expiration_date: + continue + + try: + start = parse_date(init_date) + end = parse_date(expiration_date) + except ValueError: + continue + + duration = end - start + if duration.total_seconds() <= 0: + continue + + market_value = lookup_market_value(tracking_data, symbol) if symbol else None + quantity = entry.get("quantity") + if not isinstance(quantity, (int, float)): + quantity = 0.0 + + option_price = None + if isinstance(market_value, (int, float)) and quantity != 0: + try: + option_price = abs(market_value) / abs(quantity) + except Exception: + option_price = market_value + else: + option_price = market_value if isinstance(market_value, (int, float)) else 0.0 + + cost_basis = entry.get("cost_basis") + bar_color = "lime" if isinstance(cost_basis, (int, float)) and isinstance(market_value, (int, float)) and market_value > cost_basis else "red" + + label_parts = [underlying, str(strike) if strike is not None else "", put_call] + label = " ".join(part for part in label_parts if part).strip() + if not label: + label = symbol or "option" + + grouped_options.setdefault(underlying, []).append({ + "label": label, + "symbol": symbol, + "start": start, + "duration": duration, + "bar_color": bar_color, + "cost_basis": cost_basis, + "label_color": "green" if put_call == "CALL" else "red" if put_call == "PUT" else "white", + "market_value": market_value if isinstance(market_value, (int, float)) else 0.0, + "option_price": option_price, + "quantity": quantity, + "strike": strike, + "put_call": put_call, + "underlying": underlying, + }) + + for option_list in grouped_options.values(): + option_list.sort(key=lambda opt: opt.get("option_price") or 0.0, reverse=True) + + ordered_underlyings = sorted( + grouped_options.items(), + key=lambda item: sum(opt.get("market_value", 0.0) for opt in item[1]), + reverse=True, + ) + + return ordered_underlyings + + +def _build_values_table_html(groups: List[Tuple[str, List[Dict[str, Any]]]]) -> str: + html = [ + "
", + f"", + "", + "", + "", + "", + "", + "", + "", + "", + "", + "", + ] + tracking = load_tracking() + + for underlying, options in groups: + n = len(options) + # representative mark: use tracking mark for the underlying if available + rep_mark = None + try: + entry = tracking.get(underlying) if isinstance(tracking, dict) else None + if isinstance(entry, dict): + rep_mark = entry.get("mark") + except Exception: + rep_mark = None + + underlying_label = underlying + if rep_mark is not None: + try: + underlying_label = f"{underlying} {float(rep_mark):,.2f}" + except Exception: + underlying_label = f"{underlying} {rep_mark}" + + for idx, option in enumerate(options): + values = _compute_option_values(option, tracking) + intrinsic = values["intrinsic"] + extrinsic = values["extrinsic"] + intrinsic_html = _format_money_html(intrinsic) + extrinsic_html = _format_money_html(extrinsic) + intrinsic_color = _money_color(intrinsic) + extrinsic_color = _money_color(extrinsic) + if idx == 0: + # Render underlying cell with rowspan equal to number of options + # Position cell: prevent wrapping so PUT/CALL stays on same line + html.append( + "" + "" + "" + "" + "" + "" + "" % ( + n, + underlying_label, + option["label"], + option["quantity"], + intrinsic_color, + intrinsic_html, + extrinsic_color, + extrinsic_html, + ) + ) + else: + html.append( + "" + "" + "" + "" + "" + "" % ( + option["label"], + option["quantity"], + intrinsic_color, + intrinsic_html, + extrinsic_color, + extrinsic_html, + ) + ) + html.append("
UnderlyingPositionQtyIntrinsicExtrinsic
%s%s%s%s%s
%s%s%s%s
") + return "".join(html) + + def parse_date(value: Any) -> datetime.datetime: if isinstance(value, datetime.datetime): return value @@ -97,74 +345,53 @@ def make_gantt_chart() -> None: positions_data = load_positions() tracking_data = load_tracking() - grouped_options: Dict[str, List[Dict[str, Any]]] = {} - - for entry in positions_data.get("summary", []): - if not isinstance(entry, dict): - continue - - instrument = entry.get("instrument") or {} - symbol = entry.get("symbol") or instrument.get("symbol") - underlying = instrument.get("underlyingSymbol") or entry.get("underlying_symbol") or "UNKNOWN" - strike = entry.get("strike_price") or instrument.get("strikePrice") - put_call = (instrument.get("putCall") or instrument.get("type") or "").upper() - init_date = entry.get("init") - expiration_date = entry.get("expiration_date") - - if not init_date or not expiration_date: - continue - - try: - start = parse_date(init_date) - end = parse_date(expiration_date) - except ValueError: - continue - - duration = end - start - if duration.total_seconds() <= 0: - continue - - market_value = lookup_market_value(tracking_data, symbol) if symbol else None - cost_basis = entry.get("cost_basis") - bar_color = "lime" if isinstance(cost_basis, (int, float)) and isinstance(market_value, (int, float)) and market_value > cost_basis else "red" - - label_parts = [underlying, str(strike) if strike is not None else "", put_call] - label = " ".join(part for part in label_parts if part).strip() - if not label: - label = symbol or "option" - - grouped_options.setdefault(underlying, []).append({ - "label": label, - "start": start, - "duration": duration, - "bar_color": bar_color, - "label_color": "green" if put_call == "CALL" else "red" if put_call == "PUT" else "white", - "market_value": market_value if isinstance(market_value, (int, float)) else 0.0, - }) - - if not grouped_options: + ordered_underlyings = _ordered_option_groups(positions_data, tracking_data) + if not ordered_underlyings: print("No valid option positions found for Gantt chart.") return - ordered_underlyings = sorted( - grouped_options.items(), - key=lambda item: sum(opt["market_value"] for opt in item[1]), - reverse=True, - ) - rows: List[str] = [] starts: List[datetime.datetime] = [] durations: List[datetime.timedelta] = [] - colors: List[str] = [] + colors: List[Any] = [] + outline_colors: List[Optional[str]] = [] label_colors: List[str] = [] + expiration_labels: List[str] = [] for _, options in ordered_underlyings: for option in options: rows.append(option["label"]) starts.append(option["start"]) durations.append(option["duration"]) - colors.append(option["bar_color"]) + # compute opacity from market_value and cost_basis: 1 - min(1, market_value/cost_basis) + market_value = option.get("market_value") + cost_basis = option.get("cost_basis") + alpha = 1.0 + outline_color = None + try: + if isinstance(market_value, (int, float)) and isinstance(cost_basis, (int, float)) and cost_basis != 0: + ratio = market_value / cost_basis + alpha = 1.0 - min(1.0, ratio) + # Determine outline color for all bars + if isinstance(cost_basis, (int, float)) and isinstance(market_value, (int, float)) and cost_basis == market_value: + outline_color = "white" + else: + outline_color = POSITIVE_COLOR if option.get("bar_color") in ("lime", "green") else NEGATIVE_COLOR if option.get("bar_color") in ("red",) else "white" + except Exception: + alpha = 1.0 + # map base color to hex and convert to RGBA + base_hex = POSITIVE_COLOR if option.get("bar_color") in ("lime", "green") else NEGATIVE_COLOR if option.get("bar_color") in ("red",) else "#ffffff" + try: + h = base_hex.lstrip("#") + r = int(h[0:2], 16) / 255.0 + g = int(h[2:4], 16) / 255.0 + b = int(h[4:6], 16) / 255.0 + colors.append((r, g, b, max(0.0, min(1.0, alpha)))) + except Exception: + colors.append(base_hex) + outline_colors.append(outline_color) label_colors.append(option["label_color"]) + expiration_labels.append((option["start"] + option["duration"]).strftime("%m/%d/%Y")) try: mdates, plt, np = import_plotting() @@ -172,32 +399,69 @@ def make_gantt_chart() -> None: print(exc) return - fig, ax = plt.subplots(figsize=(12, max(4, len(rows) * 0.5))) + fig, ax = plt.subplots(figsize=(9, max(6, len(rows) * 0.6))) fig.patch.set_facecolor("#1e1e1e") ax.set_facecolor("#252525") + plt.rcParams["font.family"] = "DejaVu Sans" + + y_positions = np.arange(len(rows))[::-1] * 13 - y_positions = np.arange(len(rows))[::-1] * 10 for i in range(len(rows)): ax.broken_barh( [(mdates.date2num(starts[i]), durations[i].days)], - (y_positions[i] - 2.5, 5), + (y_positions[i] - 3, 6), facecolors=colors[i], ) + # Draw right-edge outlines for transparent bars + for i in range(len(rows)): + if outline_colors[i] is not None: + x_right = mdates.date2num(starts[i]) + durations[i].days + ax.plot( + [x_right, x_right], + [y_positions[i] - 3, y_positions[i] + 3], + color=outline_colors[i], + linewidth=1.5, + solid_capstyle="butt", + ) + + # intrinsic/extrinsic labels are intentionally omitted from the Gantt chart + ax.set_yticks(y_positions) ax.set_yticklabels(rows) for tick, color in zip(ax.get_yticklabels(), label_colors): tick.set_color(color) + try: + tick.set_fontsize(10) + except Exception: + pass + + x_max = max(mdates.date2num(starts[i] + durations[i]) for i in range(len(rows))) if rows else None + if x_max is not None: + current_xlim = ax.get_xlim() + right_margin = max(2.5, (x_max - current_xlim[0]) * 0.04) + ax.set_xlim(current_xlim[0], max(current_xlim[1], x_max + right_margin + 1.5)) + label_x = x_max + right_margin + 10.0 + for i, exp_label in enumerate(expiration_labels): + ax.text( + label_x, + y_positions[i], + exp_label, + color="white", + ha="left", + va="center", + fontsize=9, + ) ax.xaxis_date() ax.xaxis.set_major_formatter(mdates.DateFormatter("%Y-%m-%d")) fig.autofmt_xdate(rotation=45) - ax.tick_params(colors="white", which="both") + ax.tick_params(colors="white", which="both", labelsize=10) for spine in ax.spines.values(): spine.set_color("white") - ax.set_title("Option position Gantt chart", color="white") + ax.set_title("Option position Gantt chart", color="white", fontsize=14) plt.tight_layout() os.makedirs(DATA_DIR, exist_ok=True) @@ -207,3 +471,56 @@ def make_gantt_chart() -> None: if __name__ == "__main__": make_gantt_chart() + + +def streamlit_dashboard() -> None: + try: + import streamlit as st + except Exception as exc: + raise RuntimeError("Streamlit is required for the dashboard. Install with `pip install streamlit`.") from exc + + st.set_page_config(page_title="Options Dashboard", layout="wide") + st.markdown( + f""" + + """, + unsafe_allow_html=True, + ) + st.title("Options Dashboard") + + # Ensure chart exists (generate if needed) + try: + make_gantt_chart() + except Exception: + pass + + col1, col2 = st.columns([3, 1.65]) + + with col1: + if os.path.exists(OUTPUT_FILE): + st.image(OUTPUT_FILE, width=700) + else: + st.info("No Gantt chart available. Run the position update to generate `gantt.png`.") + + positions = [] + try: + positions_data = load_positions() + tracking_data = load_tracking() + positions = _ordered_option_groups(positions_data, tracking_data) + except Exception: + positions = [] + + with col2: + if positions: + st.markdown(_build_values_table_html(positions), unsafe_allow_html=True) + else: + st.info("No position values available") diff --git a/requirements.txt b/requirements.txt index 7079a87..c00b3ef 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,3 +2,4 @@ requests>=2.0 cryptography>=3.0 matplotlib>=3.0 numpy>=1.20 +streamlit>=1.0 diff --git a/streamlit_app.py b/streamlit_app.py new file mode 100644 index 0000000..d54e39c --- /dev/null +++ b/streamlit_app.py @@ -0,0 +1,18 @@ +# Copyright 2026 Valentin Richter + +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at + +# http://www.apache.org/licenses/LICENSE-2.0 + +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from plot import streamlit_dashboard + +if __name__ == "__main__": + streamlit_dashboard() diff --git a/test.py b/test.py index 350201e..da0beb3 100644 --- a/test.py +++ b/test.py @@ -12,44 +12,245 @@ # See the License for the specific language governing permissions and # limitations under the License. +""" +Integration tests for optchart core functionality. + +Tests verify the complete workflow: + 1. OAuth authentication and token management + 2. Fetching option positions from Schwab API + 3. Tracking option prices and underlying asset prices + 4. Generating Gantt chart visualization (gantt.png) +""" + import compileall import json import os import sys +from typing import Any, Dict, List root = os.path.dirname(os.path.abspath(__file__)) -print("Compiling Python files for syntax check...") -ok = compileall.compile_dir(root, force=True, quiet=1) -if not ok: - print("Compilation errors found.") - sys.exit(2) - -issues = False -auth_dir = os.path.join(root, "auth") -data_dir = os.path.join(root, "data") - -def check_json(path: str) -> None: - global issues - if os.path.exists(path): - try: - with open(path, "r", encoding="utf-8") as f: - json.load(f) - print(f"OK JSON: {path}") - except Exception as e: - print(f"Invalid JSON: {path} {e}") - issues = True - - -check_json(os.path.join(auth_dir, "client.json")) -check_json(os.path.join(auth_dir, "token.json")) -check_json(os.path.join(auth_dir, "token_response.json")) -check_json(os.path.join(data_dir, "positions.json")) -check_json(os.path.join(data_dir, "tracking.json")) - -if issues: - print("One or more JSON files are invalid.") - sys.exit(3) - -print("All checks passed.") -sys.exit(0) + +def check_syntax() -> bool: + """Verify all Python files compile without syntax errors.""" + print("Checking Python syntax...") + ok = compileall.compile_dir(root, force=True, quiet=1) + if not ok: + print(" [FAILED] Syntax errors found in Python files") + return False + print(" [PASS] All Python files compile successfully") + return True + + +def check_json_files() -> bool: + """Verify JSON files exist and are valid (if they exist).""" + print("Checking JSON files...") + issues = False + auth_dir = os.path.join(root, "auth") + data_dir = os.path.join(root, "data") + + json_files = [ + os.path.join(auth_dir, "client.json"), + os.path.join(auth_dir, "token.json"), + os.path.join(data_dir, "positions.json"), + os.path.join(data_dir, "tracking.json"), + ] + + for path in json_files: + if os.path.exists(path): + try: + with open(path, "r", encoding="utf-8") as f: + json.load(f) + print(f" [OK] Valid JSON: {os.path.relpath(path, root)}") + except Exception as e: + print(f" [FAILED] Invalid JSON: {os.path.relpath(path, root)} - {e}") + issues = True + + return not issues + + +def check_positions_structure() -> bool: + """Verify positions.json has the expected structure if it exists.""" + print("Checking positions data structure...") + data_dir = os.path.join(root, "data") + pos_file = os.path.join(data_dir, "positions.json") + + if not os.path.exists(pos_file): + print(" [INFO] No positions file found (expected on first run)") + return True + + try: + with open(pos_file, "r", encoding="utf-8") as f: + data = json.load(f) + + if not isinstance(data, dict): + print(" [FAILED] Positions data should be a JSON object") + return False + + if "accounts" not in data: + print(" [FAILED] Positions data missing 'accounts' key") + return False + + accounts = data["accounts"] + if not isinstance(accounts, list): + print(" [FAILED] 'accounts' should be a list") + return False + + # Check for option positions + has_options = False + for acct in accounts: + if isinstance(acct, dict) and "options" in acct: + if isinstance(acct["options"], list) and len(acct["options"]) > 0: + has_options = True + break + + if has_options: + print(" [OK] Positions file has correct structure with options") + else: + print(" [INFO] Positions file valid but contains no option positions") + + return True + + except Exception as e: + print(f" [FAILED] Error validating positions structure: {e}") + return False + + +def check_tracking_structure() -> bool: + """Verify tracking.json has the expected structure if it exists.""" + print("Checking tracking data structure...") + data_dir = os.path.join(root, "data") + track_file = os.path.join(data_dir, "tracking.json") + + if not os.path.exists(track_file): + print(" [INFO] No tracking file found (expected on first run)") + return True + + try: + with open(track_file, "r", encoding="utf-8") as f: + data = json.load(f) + + if not isinstance(data, dict): + print(" [FAILED] Tracking data should be a JSON object") + return False + + print(" [OK] Tracking file has correct structure") + return True + + except Exception as e: + print(f" [FAILED] Error validating tracking structure: {e}") + return False + + +def check_gitignore_protection() -> bool: + """Verify .gitignore protects sensitive local data folders.""" + print("Checking .gitignore protection...") + gitignore_file = os.path.join(root, ".gitignore") + if not os.path.exists(gitignore_file): + print(" [FAILED] .gitignore is missing") + return False + + try: + with open(gitignore_file, "r", encoding="utf-8") as f: + entries = [line.strip() for line in f if line.strip() and not line.strip().startswith("#")] + except Exception as e: + print(f" [FAILED] Unable to read .gitignore: {e}") + return False + + required = ["auth/", "data/"] + missing = [item for item in required if item not in entries] + if missing: + print(f" [FAILED] .gitignore missing entries: {', '.join(missing)}") + return False + + print(" [OK] .gitignore protects auth/ and data/") + return True + + +def check_gantt_generation() -> bool: + """Integration test: Verify gantt.png can be generated from existing data.""" + print("Testing Gantt chart generation...") + data_dir = os.path.join(root, "data") + pos_file = os.path.join(data_dir, "positions.json") + gantt_file = os.path.join(data_dir, "gantt.png") + + if not os.path.exists(pos_file): + print(" [INFO] Skipping Gantt test (no positions file)") + return True + + # Remove old gantt.png if present + if os.path.exists(gantt_file): + os.remove(gantt_file) + + try: + import plot + + plot.make_gantt_chart() + + if os.path.exists(gantt_file): + size = os.path.getsize(gantt_file) + print(f" [OK] Gantt chart generated successfully ({size} bytes)") + return True + else: + print(" [FAILED] Gantt chart file was not created") + return False + + except Exception as e: + print(f" [FAILED] Error generating Gantt chart: {e}") + return False + + +def main() -> int: + """Run all tests.""" + print("=" * 60) + print("optchart — Core Functionality Tests") + print("=" * 60) + print() + + results: List[bool] = [] + + results.append(check_syntax()) + print() + results.append(check_json_files()) + print() + results.append(check_positions_structure()) + print() + results.append(check_tracking_structure()) + print() + results.append(check_gitignore_protection()) + print() + results.append(check_gantt_generation()) + print() + results.append(check_streamlit_import()) + + print() + print("=" * 60) + passed = sum(results) + total = len(results) + print(f"Results: {passed}/{total} test groups passed") + + if all(results): + print("[OK] All tests passed") + print("="*60) + return 0 + else: + print("[FAILED] Some tests failed") + print("=" * 60) + return 1 + + +def check_streamlit_import() -> bool: + """Verify streamlit can be imported (installed).""" + print("Checking streamlit import...") + try: + import importlib + importlib.import_module("streamlit") + print(" [OK] Streamlit is importable") + return True + except Exception as e: + print(f" [FAILED] Streamlit import failed: {e}") + return False + + +if __name__ == "__main__": + sys.exit(main())