create dashboard, augment data in Gantt chart

.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: |

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 <repository>
+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)

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:

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 {}

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()