Add PG_CONN_STR to .env.example for host-side PostgreSQL connection

- Add db.py PostgreSQL abstraction layer (connection, upsert, table mgmt)
- Replace sqlite3 with psycopg2 in: live_candle_fetcher, resampler,
  data_fetcher, fetch_history, import_csv, indicators, base_strategy
- Sanitize table names (colons -> underscores) for PostgreSQL compat
- Replace INSERT OR REPLACE with ON CONFLICT upserts
- Replace pandas to_sql() with batch upsert_candles()
- Add scripts: resampler_loop, gap_detector, backup_runner, cron_scheduler
- Add migrate_sqlite_to_pg.py for one-time data migration
- Add Dockerfile, docker-compose.yml, supervisord.conf
- Add postgres/postgresql.conf tuned for 4GB RAM (Synology DS1513+)
- Add .dockerignore, .env.docker.example, secrets template
- Update requirements.txt (psycopg2-binary), .gitignore
- Add MIGRATION_PLAN.md with full plan and todo list

.dockerignore

@@ -0,0 +1,14 @@
+.venv/
+.git/
+_logs/
+_data/*.db
+_data/*.db-shm
+_data/*.db-wal
+__pycache__/
+*.pyc
+.temp/
+sdk/
+agents/
+secrets/
+.env.docker
+.env

.env.docker.example

@@ -0,0 +1,7 @@
+# Docker environment variables
+# Copy to .env.docker and fill in real values.
+# DO NOT commit the real .env.docker file to git.
+
+POSTGRES_PASSWORD=change_me
+PG_CONN_STR=postgresql://hyper:change_me@postgres:5432/hyper
+COINGECKO_API_KEY=

.env.example

@@ -19,6 +19,11 @@ AGENT_PRIVATE_KEY=
 # Optional: CoinGecko API key to reduce rate limits for market cap fetches
 COINGECKO_API_KEY=
 
+# PostgreSQL connection string (for host-side scripts: indicators, strategies)
+# When running in Docker, this is set in .env.docker
+# Example: PG_CONN_STR=postgresql://hyper:your_password@localhost:5432/hyper
+PG_CONN_STR=
+
 # Optional: Set a custom environment for development/testing
 # E.g., DEBUG=true
 DEBUG=

.gitignore

@@ -43,3 +43,7 @@ agents/
 .DS_Store
 Thumbs.db
 .opencode/
+
+# --- Docker ---
+secrets/
+.env.docker

Dockerfile

@@ -0,0 +1,22 @@
+FROM python:3.11-slim
+
+# Install supervisor for process management
+RUN apt-get update && apt-get install -y --no-install-recommends supervisor && \
+    rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+# Install Python dependencies
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy application source files
+COPY . .
+
+# Copy supervisord configuration
+COPY supervisord.conf /etc/supervisor/conf.d/supervisord.conf
+
+# Create required directories
+RUN mkdir -p /app/_data /app/_logs
+
+CMD ["/usr/bin/supervisord", "-c", "/etc/supervisor/conf.d/supervisord.conf"]

MIGRATION_PLAN.md

@@ -0,0 +1,126 @@
+# Migration Plan: SQLite → PostgreSQL + Docker on Synology DS1513+
+
+## Architecture Decisions
+
+| Decision | Choice | Rationale |
+|----------|--------|-----------|
+| Schema | Keep table-per-coin-timeframe (652 tables) | Minimal code changes, PostgreSQL handles it well |
+| Table names | Sanitize `:` → `_` (e.g., `xyz_BRENTOIL_1m`) | PostgreSQL compatibility |
+| Secrets | Docker env_file + bind-mount | Secure, rotate-friendly, Synology-compatible |
+| Gap detection | New `gap_detector.py` | Fills data gaps when system is down |
+| Backup | Daily `pg_dump` to shared folder | Accessible via File Station, Hyper Backup compatible |
+| Host integration | Expose PostgreSQL port 5432 | Host scripts connect to `localhost:5432` |
+| Migration | Two-phase (offline + cutover) | Minimizes downtime |
+| Legacy tables | Skip `market_cap`, `candles`, `daily` | Not used by current code |
+
+## Container Layout
+
+```
+┌─────────────────────────────────────────────────────┐
+│              Docker Compose                         │
+├─────────────────────────────────────────────────────┤
+│  ┌──────────────┐  ┌──────────────────────────────┐  │
+│  │ PostgreSQL   │  │ Data Collector (supervisord)│  │
+│  │ postgres:15- │  │ python:3.11-slim             │  │
+│  │ alpine       │  │                              │  │
+│  │              │  │ • live_candle_fetcher (cont)│  │
+│  │ shared_buff  │  │ • resampler_loop (cont)      │  │
+│  │ =128MB       │  │ • indicators_fetcher (cont) │  │
+│  │              │  │ • cron_scheduler (cont)      │  │
+│  │ Vol:pg_data  │  │   - data_fetcher (daily)     │  │
+│  │ Port:5432    │  │   - fetch_history (daily)    │  │
+│  │ exposed      │  │   - gap_detector (hourly)    │  │
+│  └──────────────┘  │   - backup_runner (daily)    │  │
+│                    └──────────────────────────────┘  │
+└─────────────────────────────────────────────────────┘
+
+Host Machine: indicators.py, base_strategy.py, main_app.py
+  → connect to localhost:5432
+```
+
+## PostgreSQL Configuration (4GB RAM)
+
+```ini
+shared_buffers = 128MB
+effective_cache_size = 512MB
+work_mem = 8MB
+maintenance_work_mem = 64MB
+max_connections = 10
+max_worker_processes = 2
+checkpoint_completion_target = 0.9
+wal_buffers = 4MB
+```
+
+## Data Migration (Two-Phase)
+
+**Phase 1 (offline)**: Stop current system → run `migrate_sqlite_to_pg.py` → 2-3 hours for 1.8GB
+
+**Phase 2 (cutover)**: Start Docker containers → update host scripts to connect to `localhost:5432`
+
+## Files to Create/Modify
+
+### New Files
+1. `db.py` — PostgreSQL abstraction layer
+2. `scripts/resampler_loop.py` — Runs resampler every minute in a loop
+3. `scripts/gap_detector.py` — Detects and fills data gaps
+4. `scripts/backup_runner.py` — Daily pg_dump with 7-day retention
+5. `scripts/cron_scheduler.py` — Schedules data_fetcher, fetch_history, gap_detector, backup
+6. `migrate_sqlite_to_pg.py` — One-time data migration
+7. `Dockerfile` — Python 3.11-slim + supervisor + psycopg2-binary
+8. `docker-compose.yml` — PostgreSQL + data-collector services
+9. `supervisord.conf` — Process management
+10. `postgres/postgresql.conf` — Tuned for 4GB RAM
+11. `.dockerignore` — Docker build context exclusions
+12. `.env.docker.example` — Docker env template
+13. `secrets/pg_password.txt.example` — PG password template
+
+### Files to Modify (7)
+1. `live_candle_fetcher.py` — `sqlite3` → `db.py`
+2. `resampler.py` — `sqlite3` → `db.py`
+3. `data_fetcher.py` — `sqlite3` → `db.py`
+4. `fetch_history.py` — `sqlite3` → `db.py`
+5. `import_csv.py` — `sqlite3` → `db.py`
+6. `indicators.py` — `sqlite3` → `psycopg2`
+7. `base_strategy.py` — `sqlite3` → `psycopg2`
+
+## TODO List
+
+### Phase 1: DB Abstraction Layer
+- [x] Create `db.py` with PostgreSQL connection, table sanitization, upsert logic
+- [x] Add `psycopg2-binary` to `requirements.txt`
+
+### Phase 2: Modify Data Collection Components
+- [ ] Modify `live_candle_fetcher.py` — replace `sqlite3.connect()` with `db.get_connection()`, `INSERT OR REPLACE` with `db.upsert_candles()`, sanitize table names
+- [ ] Modify `resampler.py` — replace `sqlite3` with `db.py`, `INSERT OR REPLACE` with `db.upsert_candles()`, `?` → `%s`
+- [ ] Modify `data_fetcher.py` — replace `sqlite3` with `db.py`, `to_sql()` → `db.upsert_candles()`
+- [ ] Modify `fetch_history.py` — replace `sqlite3` with `db.py`
+- [ ] Modify `import_csv.py` — replace `sqlite3` with `db.py`, `to_sql()` → `db.upsert_candles()`
+
+### Phase 3: New Components
+- [ ] Create `scripts/resampler_loop.py` — wraps resampler in a while loop with 60s sleep
+- [ ] Create `scripts/gap_detector.py` — detects gaps in 1m data, backfills via HTTP API
+- [ ] Create `scripts/backup_runner.py` — daily pg_dump with 7-day retention
+- [ ] Create `scripts/cron_scheduler.py` — schedules data_fetcher, fetch_history, gap_detector, backup
+
+### Phase 4: Docker Setup
+- [ ] Create `Dockerfile` (python:3.11-slim + supervisor + psycopg2-binary)
+- [ ] Create `docker-compose.yml` (postgres + data-collector services)
+- [ ] Create `supervisord.conf` (live_candle_fetcher, resampler_loop, indicators_fetcher, cron_scheduler)
+- [ ] Create `postgres/postgresql.conf` (tuned for 4GB RAM)
+- [ ] Create `.dockerignore`
+- [ ] Create `.env.docker.example`
+- [ ] Create `secrets/pg_password.txt.example`
+- [ ] Update `.gitignore`
+
+### Phase 5: Host-Side Updates
+- [ ] Modify `indicators.py` on host — connect to `localhost:5432`
+- [ ] Modify `base_strategy.py` on host — connect to `localhost:5432`
+
+### Phase 6: Migration Tool
+- [ ] Create `migrate_sqlite_to_pg.py` — reads from SQLite, writes to PostgreSQL
+
+### Phase 7: Testing & Deployment
+- [ ] Commit and push to remote
+- [ ] User clones on NAS, copies `.env` and `_data/`
+- [ ] User runs migration script
+- [ ] User starts Docker containers

WIKI/dashboard_configuration.md

@@ -0,0 +1,90 @@
+# Dashboard Configuration Guide
+
+This guide explains how to configure which tables are displayed on the live terminal dashboard.
+
+## Overview
+
+The dashboard is rendered by the `DashboardRenderer` class in `dashboard.py`. It currently supports two tables:
+
+| Table Key | Title | Description |
+|-----------|-------|-------------|
+| `market` | Market Dashboard | Live prices, best bid/ask, gap, and direction for watched coins |
+| `strategies` | Strategies | Signal, signal price, last change, timeframe, and size for each enabled strategy |
+
+Each table can be independently enabled or disabled. When only one table is visible, it takes the full terminal width. When both are visible, they split side-by-side.
+
+## Configuration
+
+### Default Visibility
+
+The default table visibility is set when `DashboardRenderer` is instantiated in `main_app.py` (`MainApp.__init__`):
+
+```python
+self.renderer = DashboardRenderer(table_visibility={
+    "market": True,
+    "strategies": False,
+})
+```
+
+By default, the **market table is enabled** and the **strategies table is disabled**.
+
+### Changing Default Visibility
+
+To change which tables are shown by default, edit the `table_visibility` dict in `main_app.py` (`MainApp.__init__`, line 349):
+
+```python
+self.renderer = DashboardRenderer(table_visibility={
+    "market": True,
+    "strategies": True,   # enable strategies table
+})
+```
+
+### Runtime Toggling
+
+Tables can be toggled at runtime through the `MainApp.toggle_table()` method, which delegates to `DashboardRenderer.toggle_table()`:
+
+```python
+# Flip the strategies table on/off
+app.toggle_table("strategies")
+
+# Explicitly enable
+app.toggle_table("strategies", enabled=True)
+
+# Explicitly disable
+app.toggle_table("strategies", enabled=False)
+```
+
+The same methods are available directly on the renderer:
+
+```python
+renderer = DashboardRenderer()
+renderer.toggle_table("market")              # flip
+renderer.toggle_table("strategies", False)   # disable
+```
+
+## How It Works
+
+### DashboardRenderer (`dashboard.py`)
+
+- `__init__(console=None, table_visibility=None)` — accepts an optional `table_visibility` dict. If not provided, defaults to `{"market": True, "strategies": False}`.
+- `toggle_table(table_name, enabled=None)` — flips the visibility state when `enabled` is `None`, or sets it to the given boolean. Raises `ValueError` for unknown table names.
+- `build_layout(...)` — conditionally builds only the tables that are enabled, then arranges them:
+  - **One table:** `Layout(table)` — full width
+  - **Two tables:** `Layout.split_row(Layout(t1), Layout(t2))` — side-by-side
+  - **Zero tables:** empty `Layout`
+
+### MainApp (`main_app.py`)
+
+- `MainApp.__init__` creates the `DashboardRenderer` with the `table_visibility` config.
+- `MainApp.toggle_table(table_name, enabled=None)` delegates to the renderer for runtime toggling.
+- `MainApp.display_dashboard()` calls `renderer.build_layout()` which respects the current visibility settings.
+
+## File Reference
+
+| File | Line | Description |
+|------|------|-------------|
+| `dashboard.py` | 22 | `DashboardRenderer.__init__` — accepts `table_visibility` parameter |
+| `dashboard.py` | 32 | `toggle_table()` method — flips or sets table visibility |
+| `dashboard.py` | 172 | `build_layout()` — conditionally includes tables based on visibility |
+| `main_app.py` | 349 | `MainApp.__init__` — sets default `table_visibility` |
+| `main_app.py` | 385 | `MainApp.toggle_table()` — runtime toggle method |

WIKI/indicators.md

@@ -0,0 +1,263 @@
+# Indicators Guide
+
+This guide explains how to configure and use the Indicators table on the live terminal dashboard.
+
+## Overview
+
+The Indicators table displays computed financial indicators (e.g., WTI/BRENT ratio, live prices, moving averages, RSI) with their current value, 1-hour and 1-day percentage changes, and deviation from a long-term average.
+
+The system is **config-driven** — new indicators are added by editing `_data/indicators.json`. No code changes are required for standard indicator types.
+
+## Dashboard Table
+
+The Indicators table is displayed below the Market table in the dashboard. It shows:
+
+| Column | Description |
+|--------|-------------|
+| `#` | Indicator number |
+| `Indicator` | Display name from config |
+| `Value` | Current indicator value |
+| `1h Change` | Percentage change over the last 1 hour |
+| `1D Change` | Percentage change over the last 1 day |
+| `Deviation` | Deviation from the long-term average |
+
+Changes are color-coded: **green** for positive, **red** for negative, **yellow** for neutral.
+
+## Configuration
+
+### Default Visibility
+
+The Indicators table is enabled by default. The visibility is set in `main_app.py` (`MainApp.__init__`):
+
+```python
+self.renderer = DashboardRenderer(table_visibility={
+    "market": True,
+    "strategies": False,
+    "indicators": True,
+})
+```
+
+### Runtime Toggling
+
+Toggle the Indicators table at runtime:
+
+```python
+app.toggle_table("indicators")              # flip on/off
+app.toggle_table("indicators", enabled=True)  # explicitly enable
+app.toggle_table("indicators", enabled=False) # explicitly disable
+```
+
+## Indicator Types
+
+The following indicator types are supported in `_data/indicators.json`:
+
+### `ratio` — A/B Ratio
+
+Computes `numerator / denominator`.
+
+```json
+"wti_brent_ratio": {
+    "display_name": "WTI/BRENT",
+    "type": "ratio",
+    "numerator": "xyz:CL",
+    "denominator": "xyz:BRENTOIL",
+    "changes": ["1h", "1d"],
+    "show_deviation": true,
+    "min_data_points": 100,
+    "fallback_reference": 0.96065
+}
+```
+
+- **Value**: `live(numerator) / live(denominator)` from latest 1m candle closes
+- **1h Change**: compares to ratio from 1h candle close prices
+- **1D Change**: compares to ratio from 1d candle close prices
+- **Deviation**: `(current - long_avg) / long_avg * 100`, where `long_avg` is the mean of daily ratios over all available history. If fewer than `min_data_points` (default 100) daily data points exist and `fallback_reference` is set, the fallback value is used instead.
+
+```json
+"gold_silver_ratio": {
+    "display_name": "GOLD/SILVER",
+    "type": "ratio",
+    "numerator": "xyz:GOLD",
+    "denominator": "xyz:SILVER",
+    "changes": ["1h", "1d"],
+    "show_deviation": true,
+    "min_data_points": 100,
+    "fallback_reference": 61.59
+}
+```
+
+### `price` — Single Price
+
+```json
+"wti_price": {
+    "display_name": "WTI",
+    "type": "price",
+    "coin": "xyz:CL",
+    "changes": ["1h", "1d"],
+    "show_deviation": true
+}
+```
+
+- **Value**: latest close price from `{coin}_1m` table
+- **1h/1D Change**: compares to close from 1h/1d candle tables
+- **Deviation**: `(current - long_avg) / long_avg * 100`, where `long_avg` is the mean of daily closes
+
+### `spread` — Price Difference
+
+Computes `numerator - denominator`.
+
+```json
+"wti_brent_spread": {
+    "display_name": "WTI-BRENT Spread",
+    "type": "spread",
+    "numerator": "xyz:CL",
+    "denominator": "xyz:BRENTOIL",
+    "changes": ["1h", "1d"],
+    "show_deviation": true
+}
+```
+
+### `diff_pct` — Percentage Difference
+
+Computes `(numerator - denominator) / denominator * 100`.
+
+```json
+"wti_brent_diff": {
+    "display_name": "WTI-BRENT Diff%",
+    "type": "diff_pct",
+    "numerator": "xyz:CL",
+    "denominator": "xyz:BRENTOIL",
+    "changes": ["1h", "1d"],
+    "show_deviation": true
+}
+```
+
+### `ma` — Moving Average
+
+```json
+"wti_ma_20": {
+    "display_name": "WTI MA(20)",
+    "type": "ma",
+    "coin": "xyz:CL",
+    "timeframe": "1h",
+    "period": 20,
+    "changes": ["1h", "1d"],
+    "show_deviation": true
+}
+```
+
+- **Value**: latest SMA value on the specified timeframe
+- **1h Change**: compares to MA value from 1h candle table
+- **1D Change**: compares to MA value from 1d candle table
+- **Deviation**: `(current_price - MA) / MA * 100` (how far the live price is from the MA)
+
+### `rsi` — Relative Strength Index
+
+```json
+"wti_rsi_14": {
+    "display_name": "WTI RSI(14)",
+    "type": "rsi",
+    "coin": "xyz:CL",
+    "timeframe": "1h",
+    "period": 14,
+    "changes": ["1h", "1d"],
+    "show_deviation": true
+}
+```
+
+- **Value**: latest RSI value (0-100) on the specified timeframe
+- **1h/1D Change**: absolute change in RSI points
+- **Deviation**: `RSI - 50` (deviation from neutral)
+
+### `custom` — Custom Function
+
+Calls a user-defined Python function.
+
+```json
+"custom_indicator": {
+    "display_name": "My Custom Indicator",
+    "type": "custom",
+    "module": "indicators.custom_indicators",
+    "function": "my_custom_calc",
+    "args": {"param1": "value1"},
+    "changes": ["1h", "1d"],
+    "show_deviation": true
+}
+```
+
+The custom function must accept `db_path` as the first argument, plus any `args` from the config, and return a dict:
+
+```python
+def my_custom_calc(db_path, **kwargs):
+    return {
+        "value": 0.96611,
+        "reference": 0.96044,
+        "changes": {"1h": 0.12, "1d": -0.45},
+        "deviation": 0.59
+    }
+}
+```
+
+## Data Sources
+
+All indicator calculations read from the SQLite database `_data/market_data.db`:
+
+- **Live value**: latest close price from `{coin}_1m` candle table (updated in real-time by `live_candle_fetcher.py`)
+- **1h change**: close price from `{coin}_1h` candle table (second-to-last completed 1h candle)
+- **1D change**: close price from `{coin}_1d` candle table (second-to-last completed 1d candle)
+- **Reference value**: mean of daily values over all available historical data. If fewer than `min_data_points` daily data points exist and `fallback_reference` is set, the fallback value is used instead.
+
+## Process Architecture
+
+```
+indicators_fetcher.py (subprocess, runs every 30s)
+    |
+    +---> indicators.py (IndicatorCalculator)
+    |         |
+    |         +---> _data/market_data.db (SQLite candle data)
+    |         +---> _data/indicators.json (config)
+    |
+    +---> _logs/indicators_status.json (output)
+            |
+            +---> main_app.py (MainApp.read_indicators_status)
+                    |
+                    +---> dashboard.py (DashboardRenderer.build_indicators_table)
+```
+
+## File Reference
+
+| File | Description |
+|------|-------------|
+| `_data/indicators.json` | Indicator definitions (config) |
+| `indicators.py` | `IndicatorCalculator` class — computation logic |
+| `indicators_fetcher.py` | Standalone script — runs in a loop, computes indicators, writes JSON |
+| `dashboard.py` | `DashboardRenderer.build_indicators_table()` — renders the table |
+| `main_app.py` | `run_indicators_fetcher()` — process target; `MainApp.read_indicators_status()` — reads JSON |
+| `_logs/indicators_status.json` | Output file with computed indicator values |
+
+## Adding a New Indicator
+
+1. Edit `_data/indicators.json` and add a new entry:
+
+```json
+"my_new_indicator": {
+    "display_name": "My Indicator",
+    "type": "price",
+    "coin": "BTC",
+    "changes": ["1h", "1d"],
+    "show_deviation": true
+}
+```
+
+2. Restart the application (`python main_app.py`). The Indicators Fetcher will automatically pick up the new config on its next run.
+
+No code changes are needed for standard indicator types (`ratio`, `price`, `spread`, `diff_pct`, `ma`, `rsi`). For custom calculations, use the `custom` type.
+
+### Optional Deviation Config Fields
+
+The following optional fields control the deviation reference value:
+
+| Field | Type | Default | Description |
+|-------|------|---------|-------------|
+| `min_data_points` | int | 100 | Minimum number of historical daily data points required before using the computed mean as the reference |
+| `fallback_reference` | float | null | If set and available data points are below `min_data_points`, this value is used as the reference instead of the computed mean |

WIKI/symbol_management.md

@@ -0,0 +1,221 @@
+# Symbol Management Guide
+
+This guide explains how to add or remove Hyperliquid trading symbols (coins) from the trading bot's dashboard, data pipeline, and market cap tracking.
+
+## Overview
+
+The system tracks coins through multiple interconnected components. Each component reads its coin list from a specific source:
+
+| Component | Source | Purpose |
+|-----------|--------|---------|
+| Dashboard display | `WATCHED_COINS` in `main_app.py` | Shows live prices in terminal |
+| Live candle fetcher | `--coins` CLI arg (from `WATCHED_COINS`) | Collects 1-minute candle data |
+| Resampler | `--coins` CLI arg (from `WATCHED_COINS`) | Resamples 1m data to 15+ timeframes |
+| Live price feed | `coins_to_watch` arg (from `WATCHED_COINS`) | WebSocket BBO/trade subscriptions |
+| Market cap fetcher | `coin_id_map.json` | CoinGecko market cap data |
+| Resampling status | `resampling_status.json` | Tracks progress per coin/timeframe |
+| Market cap summary | `market_cap_data.json` | Aggregated market cap snapshots |
+
+## Data Pipeline
+
+```
+Hyperliquid WebSocket
+    |
+    +---> Live Candle Fetcher (1m candles) --> SQLite: {coin}_1m
+    |         |
+    |         +---> Resampler --> SQLite: {coin}_{3m,5m,15m,...,1M}
+    |
+    +---> Live Price Feed (BBO/trades) --> shared_prices dict --> Dashboard
+
+CoinGecko API
+    |
+    +---> Market Cap Fetcher --> SQLite: {coin}_market_cap
+                            --> market_cap_data.json (summary)
+```
+
+All historical data is stored in `_data/market_data.db` (SQLite). Existing data is **preserved** when removing coins; only new data collection stops.
+
+---
+
+## Adding a Symbol
+
+### Step 1: Add to the Watched Coins List
+
+Edit `main_app.py` (line 23):
+
+```python
+WATCHED_COINS = ["BTC", "ETH", "SOL", "BNB", "HYPE", "NEW_COIN", "xyz:BRENTOIL", "xyz:CL"]
+```
+
+### Step 2: Add Display Name (Optional)
+
+If the symbol contains special characters or you want a custom display name, add it to `COIN_DISPLAY_NAMES` in `main_app.py` (lines 25-28):
+
+```python
+COIN_DISPLAY_NAMES = {
+    "xyz:BRENTOIL": "BRENT",
+    "xyz:CL": "WTI",
+    "NEW_COIN": "NewCoin"
+}
+```
+
+### Step 3: Add to Coin ID Map (for Market Cap)
+
+Edit `_data/coin_id_map.json` and add an entry mapping the Hyperliquid symbol to the CoinGecko ID:
+
+```json
+"NEW_COIN": "new-coin-id-on-coingecko"
+```
+
+If the coin is already in the map (e.g., it was previously fetched), skip this step.
+
+### Step 4: Add to Manual Overrides (Optional)
+
+If the CoinGecko ID is ambiguous, add it to the `manual_overrides` dictionary in `coin_id_map.py` (lines 49-61):
+
+```python
+manual_overrides = {
+    "BTC": "bitcoin",
+    "ETH": "ethereum",
+    "NEW_COIN": "new-coin-id-on-coingecko",
+    ...
+}
+```
+
+### Step 5: Restart the Application
+
+Stop all running processes, then start `main_app.py`:
+
+```bash
+python main_app.py
+```
+
+The system will automatically:
+- Create new candle tables in `market_data.db`
+- Begin collecting 1-minute candle data
+- Begin resampling to all timeframes
+- Begin collecting market cap data
+- Display the coin on the dashboard
+
+---
+
+## Removing a Symbol
+
+### Step 1: Stop All Running Processes
+
+Before making changes, stop all Python processes related to the project:
+
+```powershell
+# Find running processes
+Get-WmiObject Win32_Process | Where-Object { $_.ExecutablePath -like "*python*" -and $_.CommandLine -like "*hyper*" }
+
+# Stop them (replace PIDs with actual values)
+Stop-Process -Id <PID1>, <PID2>, ... -Force
+```
+
+### Step 2: Remove from Watched Coins List
+
+Edit `main_app.py` (line 23) and remove the coin from `WATCHED_COINS`:
+
+```python
+WATCHED_COINS = ["BTC", "ETH", "SOL", "BNB", "HYPE", "SUI", "xyz:BRENTOIL", "xyz:CL"]
+```
+
+### Step 3: Remove from Resampling Status
+
+Edit `_data/resampling_status.json` and delete the entire block for the coin, e.g.:
+
+```json
+"REMOVED_COIN": {
+    "12h": { ... },
+    "148m": { ... },
+    ...
+}
+```
+
+### Step 4: Remove from Coin ID Map
+
+Edit `_data/coin_id_map.json` and delete the entry:
+
+```json
+"REMOVED_COIN": "coingecko-id"
+```
+
+### Step 5: Remove from Market Cap Summary
+
+Edit `_data/market_cap_data.json` and delete the entry:
+
+```json
+"REMOVED_COIN_market_cap": { ... }
+```
+
+### Step 6: Remove from Manual Overrides (if present)
+
+Edit `coin_id_map.py` and remove the entry from `manual_overrides`:
+
+```python
+manual_overrides = {
+    "BTC": "bitcoin",
+    "ETH": "ethereum",
+    ...
+}
+```
+
+### Step 7: Restart the Application
+
+```bash
+python main_app.py
+```
+
+**Note:** Existing data in `_data/market_data.db` (candle tables, market cap tables) is **not deleted**. The coin's data remains available for historical analysis; only new data collection stops.
+
+---
+
+## File Reference
+
+### Core Configuration
+
+| File | Line | Description |
+|------|------|-------------|
+| `main_app.py` | 23 | `WATCHED_COINS` list - master coin list for dashboard, candle fetcher, resampler, and live feed |
+| `main_app.py` | 25-28 | `COIN_DISPLAY_NAMES` dict - maps internal symbols to display names |
+| `main_app.py` | 591-594 | `required_timeframes` list - timeframes for resampling |
+
+### Data Files
+
+| File | Description |
+|------|-------------|
+| `_data/market_data.db` | SQLite database with all candle and market cap data. Tables: `{coin}_1m`, `{coin}_{timeframe}`, `{coin}_market_cap` |
+| `_data/resampling_status.json` | Tracks `last_candle_utc` and `total_candles` per coin/timeframe |
+| `_data/coin_id_map.json` | Maps Hyperliquid symbols to CoinGecko IDs for market cap fetching |
+| `_data/market_cap_data.json` | Summary of latest market cap data per coin |
+| `_data/coin_precision.json` | All Hyperliquid coins with trade precision (reference only) |
+| `_data/strategies.json` | Trading strategy configurations (separate from watched coins) |
+
+### Scripts
+
+| File | Description |
+|------|-------------|
+| `main_app.py` | Main orchestrator - starts all processes, renders dashboard |
+| `live_candle_fetcher.py` | Collects 1-minute candles via WebSocket + historical catch-up |
+| `resampler.py` | Resamples 1m candles to multiple timeframes using pandas |
+| `live_market_utils.py` | WebSocket feed for live BBO (best bid/offer) and trade data |
+| `market_cap_fetcher.py` | Fetches daily market cap data from CoinGecko API |
+| `coin_id_map.py` | Generates `coin_id_map.json` from Hyperliquid + CoinGecko APIs |
+| `dashboard_data_fetcher.py` | Fetches account balances and positions for dashboard |
+
+---
+
+## Important Notes
+
+1. **Always stop processes before editing config files.** Running processes will overwrite changes to `resampling_status.json` and `market_data.db`.
+
+2. **Existing data is preserved.** Removing a coin from the lists stops new data collection but does not delete existing data from the SQLite database.
+
+3. **The `coin_id_map.json` is auto-generated.** Running `python coin_id_map.py` regenerates it from the Hyperliquid API. Manual overrides in `coin_id_map.py` ensure correct CoinGecko mappings.
+
+4. **Market cap fetcher is not auto-started.** The market cap fetcher process is currently disabled in `main_app.py` (line 614). It can be run manually: `python market_cap_fetcher.py`.
+
+5. **Strategy coins are separate.** Trading strategies in `_data/strategies.json` define their own coins independently of `WATCHED_COINS`. A coin can be traded by a strategy even if it's not in the watched list.
+
+6. **Special symbols.** Coins with the `xyz:` prefix (e.g., `xyz:BRENTOIL`, `xyz:CL`) are synthetic/derivative symbols on Hyperliquid. They follow the same management process as regular coins.

_data/coin_id_map.json

@@ -16,7 +16,6 @@
     "AR": "arweave",
     "ARB": "osmosis-allarb",
     "ARK": "ark-3",
-    "ASTER": "astar",
     "ATOM": "lost-bitcoin-layer",
     "AVAX": "binance-peg-avalanche",
     "AVNT": "avantis",
@@ -139,7 +138,6 @@
     "POPCAT": "popcat",
     "PROMPT": "wayfinder",
     "PROVE": "succinct",
-    "PUMP": "pump-fun",
     "PURR": "purr-2",
     "PYTH": "pyth-network",
     "RDNT": "radiant-capital",
@@ -198,7 +196,6 @@
     "XRP": "ripple",
     "YGG": "yield-guild-games",
     "YZY": "yzy",
-    "ZEC": "zcash",
     "ZEN": "zenith-3",
     "ZEREBRO": "zerebro",
     "ZETA": "zeta",

_data/indicators.json

@@ -0,0 +1,32 @@
+{
+    "wti_brent_ratio": {
+        "display_name": "WTI/BRENT",
+        "type": "ratio",
+        "numerator": "xyz:CL",
+        "denominator": "xyz:BRENTOIL",
+        "changes": ["1h", "1d"],
+        "show_deviation": true,
+        "min_data_points": 100,
+        "fallback_reference": 0.96065
+    },
+    "gold_silver_ratio": {
+        "display_name": "GOLD/SILVER",
+        "type": "ratio",
+        "numerator": "xyz:GOLD",
+        "denominator": "xyz:SILVER",
+        "changes": ["1h", "1d"],
+        "show_deviation": true,
+        "min_data_points": 100,
+        "fallback_reference": 61.59
+    },
+    "xyz100_ustech_ratio": {
+        "display_name": "XYZ100/USTECH",
+        "type": "ratio",
+        "numerator": "xyz:XYZ100",
+        "denominator": "mkts:USTECH",
+        "changes": ["1h", "1d"],
+        "show_deviation": true,
+        "min_data_points": 100,
+        "fallback_reference": 41.10
+    }
+}

_data/market_cap_data.json

@@ -84,11 +84,6 @@
         "timestamp_ms": 1762214400000,
         "market_cap": 411547691.74511635
     },
-    "ASTER_market_cap": {
-        "datetime_utc": "2025-11-04 00:00:00",
-        "timestamp_ms": 1762214400000,
-        "market_cap": 122331099.54500043
-    },
     "ATOM_market_cap": {
         "datetime_utc": "2025-11-04 00:00:00",
         "timestamp_ms": 1762214400000,
@@ -699,11 +694,6 @@
         "timestamp_ms": 1762214400000,
         "market_cap": 116187315.47981949
     },
-    "PUMP_market_cap": {
-        "datetime_utc": "2025-11-04 00:00:00",
-        "timestamp_ms": 1762214400000,
-        "market_cap": 1369591728.1563232
-    },
     "PURR_market_cap": {
         "datetime_utc": "2025-11-04 00:00:00",
         "timestamp_ms": 1762214400000,
@@ -994,11 +984,6 @@
         "timestamp_ms": 1762214400000,
         "market_cap": 49793986.29032182
     },
-    "ZEC_market_cap": {
-        "datetime_utc": "2025-11-04 00:00:00",
-        "timestamp_ms": 1762214400000,
-        "market_cap": 6917445577.244665
-    },
     "ZEN_market_cap": {
         "datetime_utc": "2025-11-04 00:00:00",
         "timestamp_ms": 1762214400000,

base_strategy.py

@@ -4,7 +4,7 @@ import json
 import os
 import logging
 from datetime import datetime, timezone
-import sqlite3
+import psycopg2
 import multiprocessing
 import time
 
@@ -27,7 +27,7 @@ class BaseStrategy(ABC):
         
         self.coin = params.get("coin", "N/A")
         self.timeframe = params.get("timeframe", "N/A")
-        self.db_path = os.path.join("_data", "market_data.db")
+        self.db_path = os.environ.get("PG_CONN_STR", "postgresql://hyper:hyper@localhost:5432/hyper")
         self.status_file_path = os.path.join("_data", f"strategy_status_{self.strategy_name}.json")
 
         self.current_signal = "INIT"
@@ -38,19 +38,23 @@ class BaseStrategy(ABC):
 
     def load_data(self) -> pd.DataFrame:
         """Loads historical data for the configured coin and timeframe."""
-        table_name = f"{self.coin}_{self.timeframe}"
+        table_name = f"{self.coin.replace(':', '_')}_{self.timeframe}"
         
         periods = [v for k, v in self.params.items() if 'period' in k or '_ma' in k or 'slow' in k or 'fast' in k]
         limit = max(periods) + 50 if periods else 500
 
         try:
-            with sqlite3.connect(f"file:{self.db_path}?mode=ro", uri=True) as conn:
+            conn = psycopg2.connect(self.db_path)
+            conn.set_session(readonly=True)
+            try:
                 query = f'SELECT * FROM "{table_name}" ORDER BY datetime_utc DESC LIMIT {limit}'
                 df = pd.read_sql(query, conn, parse_dates=['datetime_utc'])
                 if df.empty: return pd.DataFrame()
                 df.set_index('datetime_utc', inplace=True)
                 df.sort_index(inplace=True)
                 return df
+            finally:
+                conn.close()
         except Exception as e:
             logging.error(f"Failed to load data from table '{table_name}': {e}")
             return pd.DataFrame()

coin_id_map.py

@@ -52,9 +52,6 @@ def update_coin_mapping():
         "SOL": "solana",
         "BNB": "binancecoin",
         "HYPE": "hyperliquid",
-        "PUMP": "pump-fun",
-        "ASTER": "astar",
-        "ZEC": "zcash",
         "SUI": "sui",
         "ACE": "endurance",
         # Add other important ones you watch here

dashboard.py

@@ -0,0 +1,347 @@
+"""
+Dashboard rendering module using rich.
+Provides DashboardRenderer for building rich terminal tables and layouts.
+"""
+
+from datetime import datetime, timezone
+
+try:
+    from rich.console import Console
+    from rich.table import Table
+    from rich.live import Live
+    from rich.layout import Layout
+    from rich.text import Text
+    from rich.padding import Padding
+    RICH_AVAILABLE = True
+except ImportError:
+    RICH_AVAILABLE = False
+
+
+class DashboardRenderer:
+    """Encapsulates all rich-based dashboard rendering logic."""
+
+    def __init__(self, console=None, table_visibility=None):
+        if not RICH_AVAILABLE:
+            raise ImportError("rich is not available. Install with: pip install rich")
+        self.console = console or Console()
+        self.previous_prices = {}
+        self.table_visibility = table_visibility or {
+            "market": True,
+            "strategies": False,
+            "indicators": True,
+            "balances": True,
+        }
+
+    def toggle_table(self, table_name, enabled=None):
+        """Toggle a table's visibility on the dashboard.
+
+        Args:
+            table_name: The key of the table to toggle (e.g. "market", "strategies").
+            enabled: If None, flips the current state. Otherwise sets to the given value.
+
+        Returns:
+            The new visibility state for the table.
+        """
+        if table_name not in self.table_visibility:
+            raise ValueError(f"Unknown table: {table_name}")
+        if enabled is None:
+            self.table_visibility[table_name] = not self.table_visibility[table_name]
+        else:
+            self.table_visibility[table_name] = enabled
+        return self.table_visibility[table_name]
+
+    def _format_price(self, price_val, width=10):
+        """Format a price value with appropriate precision."""
+        try:
+            price_float = float(price_val)
+            if price_float < 1:
+                return f"{price_float:>{width}.6f}"
+            elif price_float < 100:
+                return f"{price_float:>{width}.4f}"
+            else:
+                return f"{price_float:>{width}.2f}"
+        except (ValueError, TypeError):
+            return f"{'Loading...':>{width}}"
+
+    def build_market_table(self, watched_coins, prices, display_names):
+        """Build the market dashboard table."""
+        table = Table(title="Market Dashboard", show_header=True, header_style="bold cyan", title_style="bold white")
+        table.add_column("#", justify="right", style="dim", width=3)
+        table.add_column("Coin", justify="center", width=8)
+        table.add_column("Best Bid", justify="right")
+        table.add_column("Live Price", justify="right")
+        table.add_column("Best Ask", justify="right")
+        table.add_column("Gap", justify="right")
+        table.add_column("Dir", justify="center", width=3)
+
+        for i, coin in enumerate(watched_coins, 1):
+            display_name = display_names.get(coin, coin)
+            mid = prices.get(coin)
+            bid = prices.get(f"{coin}_bid")
+            ask = prices.get(f"{coin}_ask")
+
+            formatted_mid = self._format_price(mid)
+            formatted_bid = self._format_price(bid)
+            formatted_ask = self._format_price(ask)
+
+            gap_str = "Loading..."
+            gap_style = "dim"
+            try:
+                gap_val = float(ask) - float(bid)
+                if gap_val < 1:
+                    gap_str = f"{gap_val:.6f}"
+                else:
+                    gap_str = f"{gap_val:.4f}"
+                gap_style = "green" if gap_val > 0 else "red"
+            except (ValueError, TypeError):
+                pass
+
+            direction = " "
+            direction_style = "dim"
+            prev_mid = self.previous_prices.get(coin)
+            if prev_mid is not None and mid is not None:
+                try:
+                    if float(mid) > float(prev_mid):
+                        direction = "↑"
+                        direction_style = "green"
+                    elif float(mid) < float(prev_mid):
+                        direction = "↓"
+                        direction_style = "red"
+                except (ValueError, TypeError):
+                    pass
+
+            table.add_row(
+                str(i), display_name, formatted_bid, formatted_mid, formatted_ask,
+                Text(gap_str, style=gap_style),
+                Text(direction, style=direction_style)
+            )
+
+            if coin == "SUI":
+                table.add_section()
+
+            if mid is not None:
+                self.previous_prices[coin] = mid
+
+        return table
+
+    def build_strategy_table(self, strategy_statuses, strategy_configs):
+        """Build the strategies table."""
+        table = Table(title="Strategies", show_header=True, header_style="bold cyan", title_style="bold white")
+        table.add_column("#", justify="center", width=3)
+        table.add_column("Strategy Name", width=25)
+        table.add_column("Coin", justify="center", width=8)
+        table.add_column("Signal", justify="center", width=10)
+        table.add_column("Signal Price", justify="right", width=14)
+        table.add_column("Last Change", justify="right", width=19)
+        table.add_column("TF", justify="center", width=7)
+        table.add_column("Size", justify="center", width=10)
+
+        for i, (name, status) in enumerate(strategy_statuses.items(), 1):
+            signal = status.get('current_signal', 'N/A')
+            price = status.get('signal_price')
+            price_display = f"{price:.4f}" if isinstance(price, (int, float)) else "-"
+            last_change = status.get('last_signal_change_utc')
+            last_change_display = 'Never'
+            if last_change:
+                dt_utc = datetime.fromisoformat(last_change.replace('Z', '+00:00')).replace(tzinfo=timezone.utc)
+                dt_local = dt_utc.astimezone(None)
+                last_change_display = dt_local.strftime('%Y-%m-%d %H:%M')
+
+            config_params = strategy_configs.get(name, {}).get('parameters', {})
+            coin = status.get('coin', config_params.get('coin', 'N/A'))
+
+            size = status.get('size')
+            if not size:
+                if 'coins_to_copy' in config_params:
+                    size = 'Multi'
+                else:
+                    size = config_params.get('size', 'N/A')
+
+            timeframe = config_params.get('timeframe', 'N/A')
+
+            signal_style = ""
+            if signal == "BUY":
+                signal_style = "green"
+            elif signal == "SELL":
+                signal_style = "red"
+            elif signal == "NEUTRAL":
+                signal_style = "yellow"
+
+            table.add_row(
+                str(i), name, coin,
+                Text(signal, style=signal_style) if signal_style else signal,
+                price_display, last_change_display, timeframe, str(size)
+            )
+
+        return table
+
+    def _format_change_value(self, value):
+        """Format a percentage change value with color styling."""
+        if value is None:
+            return Text("N/A", style="dim")
+        if value > 0:
+            return Text(f"+{value:.2f}%", style="green")
+        elif value < 0:
+            return Text(f"{value:.2f}%", style="red")
+        else:
+            return Text(f"{value:.2f}%", style="yellow")
+
+    def _format_value(self, value, width=12):
+        """Format a numeric value for display."""
+        if value is None:
+            return Text("N/A", style="dim")
+        try:
+            val = float(value)
+            if abs(val) < 1:
+                return Text(f"{val:>{width}.6f}")
+            elif abs(val) < 100:
+                return Text(f"{val:>{width}.4f}")
+            else:
+                return Text(f"{val:>{width}.2f}")
+        except (ValueError, TypeError):
+            return Text("N/A", style="dim")
+
+    def build_indicators_table(self, indicators_status):
+        """Build the indicators dashboard table."""
+        table = Table(title="Indicators", show_header=True, header_style="bold cyan", title_style="bold white")
+        table.add_column("#", justify="right", style="dim", width=3)
+        table.add_column("Indicator", width=20)
+        table.add_column("Value", justify="right")
+        table.add_column("1h Change", justify="right", width=12)
+        table.add_column("1D Change", justify="right", width=12)
+        table.add_column("Deviation", justify="right", width=12)
+
+        if not indicators_status:
+            table.add_row("1", "Loading...", "N/A", "N/A", "N/A", "N/A")
+            return table
+
+        indicators = indicators_status.get("indicators", {})
+        for i, (name, data) in enumerate(indicators.items(), 1):
+            display_name = data.get("display_name", name)
+            value = data.get("value")
+            changes = data.get("changes", {})
+            deviation = data.get("deviation")
+
+            formatted_value = self._format_value(value)
+            change_1h = self._format_change_value(changes.get("1h"))
+            change_1d = self._format_change_value(changes.get("1d"))
+
+            if deviation is not None:
+                if deviation > 0:
+                    deviation_str = Text(f"+{deviation:.2f}%", style="green")
+                elif deviation < 0:
+                    deviation_str = Text(f"{deviation:.2f}%", style="red")
+                else:
+                    deviation_str = Text(f"{deviation:.2f}%", style="yellow")
+            else:
+                deviation_str = Text("N/A", style="dim")
+
+            table.add_row(
+                str(i), display_name, formatted_value,
+                change_1h, change_1d, deviation_str
+            )
+
+        return table
+
+    def build_balances_table(self, account_data, prices=None):
+        """Build a combined balances and open positions table.
+
+        Args:
+            account_data: dict with keys:
+                - spot_balances: list of {coin, total}
+                - positions: list of position dicts with position data
+                - account_value: float
+                - margin_used: float
+                - utilization: float
+            prices: dict mapping coin names to current mark prices
+        """
+        if prices is None:
+            prices = {}
+        table = Table(show_header=True, header_style="bold cyan", title="Account Summary")
+        table.add_column("Type", justify="center", width=8)
+        table.add_column("Coin", justify="center", width=8)
+        table.add_column("Size", justify="right", width=12)
+        table.add_column("Value", justify="right", width=12)
+
+        spot_balances = account_data.get('spot_balances', [])
+        for bal in spot_balances:
+            total = float(bal.get('total', 0))
+            if total > 0:
+                coin = bal.get('coin', 'Unknown')
+                mark_price = float(prices.get(coin, 0))
+                usd_value = total * mark_price
+                table.add_row(
+                    Text("Spot", style="blue"),
+                    coin,
+                    f"{total:,.4f}",
+                    f"${usd_value:,.2f}"
+                )
+
+        positions = account_data.get('positions', [])
+        for pos in positions:
+            position = pos.get('position', {})
+            coin = position.get('coin', 'Unknown')
+            size = float(position.get('szi', 0))
+            if size != 0:
+                position_value = float(position.get('positionValue', 0))
+                side = "LONG" if size > 0 else "SHORT"
+                side_style = "green" if size > 0 else "red"
+
+                table.add_row(
+                    Text(f"P({side})", style=side_style),
+                    coin,
+                    f"{size:,.4f}",
+                    f"${position_value:,.2f}"
+                )
+
+        if not spot_balances and not positions:
+            table.add_row("None", "-", "-", "-")
+
+        account_value = account_data.get('account_value', 0)
+        margin_used = account_data.get('margin_used', 0)
+        utilization = account_data.get('utilization', 0)
+
+# table.add_section()
+        table.add_row(
+            Text("Acct", style="bold"),
+            "-", "-",
+            f"${account_value:,.2f}"
+        )
+        table.add_row(
+            Text("Util", style="bold"),
+            "-", "-",
+            f"{utilization:.2f}%"
+        )
+
+        return table
+
+    def build_layout(self, watched_coins, prices, display_names, strategy_statuses, strategy_configs, indicators_status=None, account_data=None):
+        """Build the complete dashboard layout in a 2x2 grid."""
+        from rich.layout import Layout as RichLayout
+
+        tables = []
+
+        if self.table_visibility.get("market", True):
+            tables.append(self.build_market_table(watched_coins, prices, display_names))
+        if self.table_visibility.get("indicators", True):
+            tables.append(Padding(self.build_indicators_table(indicators_status), (0, 0, 0, 2)))
+        if account_data is not None and self.table_visibility.get("balances", True):
+            tables.append(Padding(self.build_balances_table(account_data, prices), (0, 0, 0, 2)))
+        if self.table_visibility.get("strategies", True):
+            tables.append(self.build_strategy_table(strategy_statuses, strategy_configs))
+
+        if not tables:
+            return RichLayout()
+
+        if len(tables) <= 2:
+            layout = RichLayout()
+            layout.split_row(*tables)
+            return layout
+
+        top = RichLayout(ratio=1)
+        bottom = RichLayout(ratio=2)
+        top.split_row(*tables[:2])
+        bottom.split_row(*tables[2:])
+        layout = RichLayout()
+        layout.split_column(top, bottom)
+        return layout

data_fetcher.py

@@ -4,7 +4,7 @@ import logging
 import os
 import sys
 import time
-import sqlite3
+import db
 import pandas as pd
 from datetime import datetime, timedelta, timezone
 
@@ -26,7 +26,7 @@ class CandleFetcherDB:
         self.coins = self._resolve_coins(coins_to_fetch)
         self.interval = interval
         self.days_back = days_back
-        self.db_path = os.path.join("_data", "market_data.db")
+        self.db_path = os.environ.get("PG_CONN_STR", "postgresql://hyper:hyper@localhost:5432/hyper")
         self.column_rename_map = {
             't': 'timestamp_ms', 'o': 'open', 'h': 'high', 'l': 'low', 'c': 'close', 'v': 'volume', 'n': 'number_of_trades'
         }
@@ -47,8 +47,7 @@ class CandleFetcherDB:
 
     def run(self):
         """Starts the data fetching process and reports status after each coin."""
-        with sqlite3.connect(self.db_path, timeout=10) as self.conn:
+        self.conn = db.get_connection()
-            self.conn.execute("PRAGMA journal_mode=WAL;")
         for coin in self.coins:
             logging.info(f"--- Starting process for {coin} ---")
             num_updated = self._update_data_for_coin(coin)
@@ -73,11 +72,11 @@ class CandleFetcherDB:
 
     def _get_start_time(self, coin: str) -> (int, bool):
         """Checks the database for an existing table and returns the last timestamp."""
-        table_name = f"{coin}_{self.interval}"
+        table_name = db.sanitize_table_name(coin, self.interval)
         try:
             cursor = self.conn.cursor()
-            cursor.execute(f"SELECT name FROM sqlite_master WHERE type='table' AND name='{table_name}';")
+            cursor.execute("SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = %s)", (table_name,))
-            if cursor.fetchone():
+            if cursor.fetchone()[0]:
                 query = f'SELECT MAX(timestamp_ms) FROM "{table_name}"'
                 last_ts = pd.read_sql(query, self.conn).iloc[0, 0]
                 if pd.notna(last_ts):
@@ -150,23 +149,28 @@ class CandleFetcherDB:
         return None
 
     def _save_to_sqlite_with_pandas(self, df: pd.DataFrame, coin: str, is_append: bool) -> int:
-        """Saves a pandas DataFrame to an SQLite table and returns the number of saved rows."""
+        """Saves a pandas DataFrame to a PostgreSQL table and returns the number of saved rows."""
-        table_name = f"{coin}_{self.interval}"
+        table_name = db.sanitize_table_name(coin, self.interval)
         try:
             df.rename(columns=self.column_rename_map, inplace=True)
             df['datetime_utc'] = pd.to_datetime(df['timestamp_ms'], unit='ms')
             final_df = df[['datetime_utc', 'timestamp_ms', 'open', 'high', 'low', 'close', 'volume', 'number_of_trades']]
 
-            write_mode = 'append' if is_append else 'replace'
+            if not is_append:
-            final_df.to_sql(table_name, self.conn, if_exists=write_mode, index=False)
+                # Drop and recreate the table for 'replace' mode
+                with self.conn.cursor() as cur:
+                    cur.execute(f'DROP TABLE IF EXISTS "{table_name}"')
+                    self.conn.commit()
+                db.create_candle_table(self.conn, table_name)
 
-            self.conn.execute(f'CREATE INDEX IF NOT EXISTS "idx_{table_name}_time" ON "{table_name}"(datetime_utc);')
+            records = list(final_df.itertuples(index=False, name=None))
+            db.upsert_candles(self.conn, table_name, records)
             
             num_saved = len(final_df)
             logging.info(f"Successfully saved {num_saved} candles to table '{table_name}'")
             return num_saved
         except Exception as e:
-            logging.error(f"Failed to write to SQLite table '{table_name}': {e}")
+            logging.error(f"Failed to write to table '{table_name}': {e}")
             return 0
 
 
@@ -175,7 +179,7 @@ if __name__ == "__main__":
     parser.add_argument(
         "--coins",
         nargs='+',
-        default=["BTC", "ETH", "xyz:BRENTOIL", "xyz:CL"],
+        default=["BTC", "ETH", "xyz:BRENTOIL", "xyz:CL", "xyz:GOLD", "xyz:SILVER"],
         help="List of coins to fetch (e.g., BTC ETH), or 'all' to fetch all coins."
     )
     parser.add_argument("--interval", default="1m", help="Candle interval (e.g., 1m, 5m, 1h).")

db.py

@@ -0,0 +1,132 @@
+"""
+PostgreSQL database abstraction layer for the Hyperliquid trading toolkit.
+
+Provides a thin wrapper around psycopg2 to centralize database operations,
+handle table name sanitization, and abstract SQL dialect differences
+from the SQLite-based codebase.
+"""
+
+import os
+import psycopg2
+from psycopg2.extras import execute_values
+
+PG_CONN_STR = os.environ.get(
+    "PG_CONN_STR",
+    "postgresql://hyper:hyper@localhost:5432/hyper"
+)
+
+
+def get_connection():
+    """Return a new psycopg2 connection to the PostgreSQL database."""
+    return psycopg2.connect(PG_CONN_STR)
+
+
+def sanitize_table_name(coin, timeframe):
+    """
+    Sanitize a coin/timeframe pair into a PostgreSQL-safe table name.
+
+    Replaces colons with underscores (e.g., 'xyz:BRENTOIL' -> 'xyz_BRENTOIL')
+    to ensure compatibility with PostgreSQL identifier rules.
+    """
+    return f"{coin.replace(':', '_')}_{timeframe}"
+
+
+def create_candle_table(conn, table_name):
+    """
+    Create a candle table if it does not already exist.
+
+    Schema matches the original SQLite layout:
+      datetime_utc, timestamp_ms (PK), open, high, low, close, volume, number_of_trades
+    Also creates an index on datetime_utc for time-range queries.
+    """
+    with conn.cursor() as cur:
+        cur.execute(f'''
+            CREATE TABLE IF NOT EXISTS "{table_name}" (
+                datetime_utc TIMESTAMP,
+                timestamp_ms BIGINT PRIMARY KEY,
+                open REAL,
+                high REAL,
+                low REAL,
+                close REAL,
+                volume REAL,
+                number_of_trades INTEGER
+            )
+        ''')
+        cur.execute(
+            f'CREATE INDEX IF NOT EXISTS "idx_{table_name}_time" ON "{table_name}"(datetime_utc)'
+        )
+        conn.commit()
+
+
+def upsert_candles(conn, table_name, records):
+    """
+    Batch upsert candle records using PostgreSQL ON CONFLICT.
+
+    Args:
+        conn: psycopg2 connection
+        table_name: sanitized table name (e.g., 'BTC_1m')
+        records: list of tuples (datetime_utc, timestamp_ms, open, high,
+                 low, close, volume, number_of_trades)
+
+    Returns:
+        Number of records upserted.
+    """
+    if not records:
+        return 0
+
+    with conn.cursor() as cur:
+        execute_values(
+            cur,
+            f'''
+                INSERT INTO "{table_name}"
+                (datetime_utc, timestamp_ms, open, high, low, close, volume, number_of_trades)
+                VALUES %s
+                ON CONFLICT (timestamp_ms) DO UPDATE SET
+                    datetime_utc = EXCLUDED.datetime_utc,
+                    open = EXCLUDED.open,
+                    high = EXCLUDED.high,
+                    low = EXCLUDED.low,
+                    close = EXCLUDED.close,
+                    volume = EXCLUDED.volume,
+                    number_of_trades = EXCLUDED.number_of_trades
+            ''',
+            records,
+            page_size=1000
+        )
+        conn.commit()
+    return len(records)
+
+
+def get_last_timestamp(conn, table_name):
+    """Return the most recent timestamp_ms from a table, or None."""
+    with conn.cursor() as cur:
+        cur.execute(f'SELECT MAX(timestamp_ms) FROM "{table_name}"')
+        result = cur.fetchone()
+        return result[0] if result and result[0] is not None else None
+
+
+def get_table_count(conn, table_name):
+    """Return the total row count of a table."""
+    with conn.cursor() as cur:
+        cur.execute(f'SELECT COUNT(*) FROM "{table_name}"')
+        return cur.fetchone()[0]
+
+
+def table_exists(conn, table_name):
+    """Check if a table exists in the database."""
+    with conn.cursor() as cur:
+        cur.execute(
+            "SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = %s)",
+            (table_name,)
+        )
+        return cur.fetchone()[0]
+
+
+def get_table_columns(conn, table_name):
+    """Return a list of column names for a table."""
+    with conn.cursor() as cur:
+        cur.execute(
+            "SELECT column_name FROM information_schema.columns WHERE table_name = %s",
+            (table_name,)
+        )
+        return [row[0] for row in cur.fetchall()]

docker-compose.yml

@@ -0,0 +1,42 @@
+version: "3.8"
+
+services:
+  postgres:
+    image: postgres:15-alpine
+    container_name: hyper_pg
+    restart: unless-stopped
+    environment:
+      POSTGRES_DB: hyper
+      POSTGRES_USER: hyper
+      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
+    volumes:
+      - pg_data:/var/lib/postgresql/data
+      - ./postgres/postgresql.conf:/etc/postgresql/postgresql.conf
+    command: postgres -c config_file=/etc/postgresql/postgresql.conf
+    ports:
+      - "5432:5432"
+    networks:
+      - hyper_net
+
+  data-collector:
+    build: .
+    container_name: hyper_data
+    restart: unless-stopped
+    depends_on:
+      - postgres
+    env_file:
+      - .env.docker
+    volumes:
+      - ./_data:/app/_data
+      - ./_logs:/app/_logs
+      - ./secrets:/app/secrets
+      - /volume1/docker/hyper/backups:/backups
+    networks:
+      - hyper_net
+
+volumes:
+  pg_data:
+
+networks:
+  hyper_net:
+    driver: bridge

fetch_history.py

@@ -0,0 +1,83 @@
+import requests
+import json
+import db
+import time
+from datetime import datetime, timezone
+
+DB_PATH = os.environ.get("PG_CONN_STR", "postgresql://hyper:hyper@localhost:5432/hyper")
+URL = "https://api.hyperliquid.xyz/info"
+
+def fetch_historical_candles(coin, start_ms, end_ms, interval="1m"):
+    """Fetch historical candles using the raw HTTP API."""
+    candles = []
+    current_start = start_ms
+    while current_start < end_ms:
+        payload = {
+            "type": "candleSnapshot",
+            "req": {
+                "coin": coin,
+                "interval": interval,
+                "startTime": current_start,
+                "endTime": end_ms
+            }
+        }
+        resp = requests.post(URL, json=payload)
+        batch = resp.json()
+        if not batch:
+            break
+        for candle in batch:
+            candle['coin'] = coin
+            candles.append(candle)
+        last_ts = batch[-1]['t']
+        if last_ts < current_start:
+            break
+        current_start = last_ts + 1
+        time.sleep(0.5)
+    return candles
+
+def write_candles_to_db(coin, candles, interval="1m"):
+    """Write candles to the database."""
+    table_name = db.sanitize_table_name(coin, interval)
+    conn = db.get_connection()
+    db.create_candle_table(conn, table_name)
+    records = []
+    for candle in candles:
+        record = (
+            datetime.fromtimestamp(candle['t'] / 1000, tz=timezone.utc).strftime('%Y-%m-%d %H:%M:%S'),
+            candle['t'],
+            candle.get('o'), candle.get('h'), candle.get('l'), candle.get('c'),
+            candle.get('v'), candle.get('n')
+        )
+        records.append(record)
+    db.upsert_candles(conn, table_name, records)
+    conn.close()
+
+def get_last_timestamp(coin):
+    """Get the most recent timestamp from the database."""
+    table_name = db.sanitize_table_name(coin, "1m")
+    conn = db.get_connection()
+    try:
+        return db.get_last_timestamp(conn, table_name)
+    except:
+        return None
+    finally:
+        conn.close()
+
+coins = ["mkts:USTECH", "xyz:XYZ100"]
+now_ms = int(time.time() * 1000)
+seven_days_ms = 7 * 24 * 60 * 60 * 1000
+
+for coin in coins:
+    for tf in ["1m", "1d"]:
+        start_ts = now_ms - seven_days_ms
+        if start_ts >= now_ms:
+            print(f"{coin} ({tf}): Already up to date")
+            continue
+
+        print(f"{coin} ({tf}): Fetching historical candles from {datetime.fromtimestamp(start_ts/1000, tz=timezone.utc)} to {datetime.fromtimestamp(now_ms/1000, tz=timezone.utc)}...")
+        candles = fetch_historical_candles(coin, start_ts, now_ms, interval=tf)
+        print(f"{coin} ({tf}): Fetched {len(candles)} candles")
+        write_candles_to_db(coin, candles, interval=tf)
+        print(f"{coin} ({tf}): Written to database")
+
+print("Done!")

import_csv.py

@@ -2,7 +2,7 @@ import argparse
 import logging
 import os
 import sys
-import sqlite3
+import db
 import pandas as pd
 from datetime import datetime
 
@@ -24,8 +24,8 @@ class CsvImporter:
         self.csv_path = csv_path
         self.coin = coin
         # --- FIX: Corrected the f-string syntax for the table name ---
-        self.table_name = f"{self.coin}_1m"
+        self.table_name = db.sanitize_table_name(self.coin, "1m")
-        self.db_path = os.path.join("_data", "market_data.db")
+        self.db_path = os.environ.get("PG_CONN_STR", "postgresql://hyper:hyper@localhost:5432/hyper")
         self.column_mapping = {
             'Open time': 'datetime_utc',
             'Open': 'open',
@@ -40,9 +40,8 @@ class CsvImporter:
         """Orchestrates the entire import and verification process."""
         logging.info(f"Starting import process for '{self.coin}' from '{self.csv_path}'...")
 
-        with sqlite3.connect(self.db_path) as conn:
+        conn = db.get_connection()
-            conn.execute("PRAGMA journal_mode=WAL;")
+        try:
-            
             # 1. Get the current state of the database
             db_oldest, db_newest, initial_row_count = self._get_db_state(conn)
 
@@ -58,6 +57,8 @@ class CsvImporter:
 
             # 4. Summarize and verify the import
             self._summarize_import(initial_row_count, len(new_data_df), conn)
+        finally:
+            conn.close()
 
     def _get_db_state(self, conn) -> (datetime, datetime, int):
         """Gets the oldest and newest timestamps and total row count from the DB table."""
@@ -104,9 +105,10 @@ class CsvImporter:
         return df_filtered
 
     def _append_to_db(self, df: pd.DataFrame, conn):
-        """Appends the DataFrame to the SQLite table."""
+        """Appends the DataFrame to the database."""
         logging.info(f"Appending {len(df):,} new rows to the database...")
-        df.to_sql(self.table_name, conn, if_exists='append', index=False)
+        records = list(df.itertuples(index=False, name=None))
+        db.upsert_candles(conn, self.table_name, records)
         logging.info("Append operation complete.")
 
     def _summarize_import(self, initial_count: int, added_count: int, conn):

indicators.py

@@ -0,0 +1,446 @@
+"""
+Indicator calculation module.
+Provides IndicatorCalculator for computing various financial indicators
+from SQLite candle data, including ratios, prices, moving averages, RSI,
+and custom functions.
+"""
+
+import json
+import os
+import psycopg2
+from contextlib import closing
+import importlib
+import logging
+import pandas as pd
+import numpy as np
+
+
+class IndicatorCalculator:
+    """
+    Computes indicator values from SQLite candle data.
+    Supports ratio, price, spread, diff_pct, ma, rsi, and custom types.
+    """
+
+    def __init__(self, config_path, db_path):
+        self.config_path = config_path
+        self.db_path = db_path
+        self.config = self._load_config()
+
+    def _load_config(self):
+        """Load indicator definitions from JSON config file."""
+        try:
+            with open(self.config_path, 'r', encoding='utf-8') as f:
+                return json.load(f)
+        except (FileNotFoundError, json.JSONDecodeError) as e:
+            logging.error(f"Failed to load indicators config from '{self.config_path}': {e}")
+            return {}
+
+    def _get_latest_close(self, coin, timeframe="1m"):
+        """Get the latest close price from a candle table."""
+        table = f"{coin.replace(':', '_')}_{timeframe}"
+        try:
+            with closing(psycopg2.connect(self.db_path)) as conn:
+                result = conn.execute(
+                    f'SELECT close FROM "{table}" ORDER BY timestamp_ms DESC LIMIT 1'
+                ).fetchone()
+                return float(result[0]) if result and result[0] is not None else None
+        except Exception as e:
+            logging.debug(f"Could not get latest close for {coin} ({timeframe}): {e}")
+            return None
+
+    def _get_close_n_candles_ago(self, coin, timeframe, n=1):
+        """Get the close price from n candles ago (n=1 = most recent completed candle)."""
+        table = f"{coin.replace(':', '_')}_{timeframe}"
+        try:
+            with closing(psycopg2.connect(self.db_path)) as conn:
+                result = conn.execute(
+                    f'SELECT close FROM "{table}" ORDER BY timestamp_ms DESC LIMIT 1 OFFSET {n}'
+                ).fetchone()
+                return float(result[0]) if result and result[0] is not None else None
+        except Exception as e:
+            logging.debug(f"Could not get close {n} candles ago for {coin} ({timeframe}): {e}")
+            return None
+
+    def _get_all_closes(self, coin, timeframe="1d"):
+        """Get all close prices from a candle table, ordered by time."""
+        table = f"{coin.replace(':', '_')}_{timeframe}"
+        try:
+            with closing(psycopg2.connect(self.db_path)) as conn:
+                result = conn.execute(
+                    f'SELECT close FROM "{table}" ORDER BY timestamp_ms'
+                ).fetchall()
+                return [float(r[0]) for r in result if r[0] is not None]
+        except Exception as e:
+            logging.debug(f"Could not get all closes for {coin} ({timeframe}): {e}")
+            return []
+
+    def _get_all_ratio(self, num_coin, den_coin, timeframe="1d"):
+        """Get all ratio values (num/den) from candle tables, ordered by time."""
+        num_table = f"{num_coin.replace(':', '_')}_{timeframe}"
+        den_table = f"{den_coin.replace(':', '_')}_{timeframe}"
+        try:
+            with closing(psycopg2.connect(self.db_path)) as conn:
+                result = conn.execute(
+                    f'SELECT n.close / d.close as ratio '
+                    f'FROM "{num_table}" n '
+                    f'JOIN "{den_table}" d ON n.timestamp_ms = d.timestamp_ms '
+                    f'ORDER BY n.timestamp_ms'
+                ).fetchall()
+                return [float(r[0]) for r in result if r[0] is not None]
+        except Exception as e:
+            logging.debug(f"Could not get ratio series for {num_coin}/{den_coin} ({timeframe}): {e}")
+            return []
+
+    def _get_all_spread(self, num_coin, den_coin, timeframe="1d"):
+        """Get all spread values (num - den) from candle tables, ordered by time."""
+        num_table = f"{num_coin.replace(':', '_')}_{timeframe}"
+        den_table = f"{den_coin.replace(':', '_')}_{timeframe}"
+        try:
+            with closing(psycopg2.connect(self.db_path)) as conn:
+                result = conn.execute(
+                    f'SELECT n.close - d.close as spread '
+                    f'FROM "{num_table}" n '
+                    f'JOIN "{den_table}" d ON n.timestamp_ms = d.timestamp_ms '
+                    f'ORDER BY n.timestamp_ms'
+                ).fetchall()
+                return [float(r[0]) for r in result if r[0] is not None]
+        except Exception as e:
+            logging.debug(f"Could not get spread series for {num_coin}/{den_coin} ({timeframe}): {e}")
+            return []
+
+    def _get_all_diff_pct(self, num_coin, den_coin, timeframe="1d"):
+        """Get all percentage difference values ((num-den)/den*100) from candle tables."""
+        num_table = f"{num_coin.replace(':', '_')}_{timeframe}"
+        den_table = f"{den_coin.replace(':', '_')}_{timeframe}"
+        try:
+            with closing(psycopg2.connect(self.db_path)) as conn:
+                result = conn.execute(
+                    f'SELECT (n.close - d.close) / d.close * 100 as diff_pct '
+                    f'FROM "{num_table}" n '
+                    f'JOIN "{den_table}" d ON n.timestamp_ms = d.timestamp_ms '
+                    f'ORDER BY n.timestamp_ms'
+                ).fetchall()
+                return [float(r[0]) for r in result if r[0] is not None]
+        except Exception as e:
+            logging.debug(f"Could not get diff_pct series for {num_coin}/{den_coin} ({timeframe}): {e}")
+            return []
+
+    def _compute_ma(self, closes, period):
+        """Compute Simple Moving Average using pandas."""
+        if len(closes) < period:
+            return []
+        series = pd.Series(closes)
+        ma = series.rolling(window=period).mean()
+        return ma.dropna().tolist()
+
+    def _compute_rsi(self, closes, period):
+        """Compute RSI using Wilder's smoothing method."""
+        if len(closes) < period + 1:
+            return []
+        series = pd.Series(closes)
+        delta = series.diff()
+        gain = delta.where(delta > 0, 0)
+        loss = (-delta).where(delta < 0, 0)
+        avg_gain = gain.rolling(window=period, min_periods=period).mean()
+        avg_loss = loss.rolling(window=period, min_periods=period).mean()
+        rs = avg_gain / avg_loss.replace(0, np.nan)
+        rsi = 100 - (100 / (1 + rs))
+        return rsi.dropna().tolist()
+
+    def _get_ma_value(self, coin, timeframe, period, n_candles_ago=0):
+        """Get MA value from n candles ago (0 = latest, 1 = second-to-last)."""
+        closes = self._get_all_closes(coin, timeframe)
+        if not closes:
+            return None
+        ma_values = self._compute_ma(closes, period)
+        if not ma_values:
+            return None
+        if n_candles_ago < len(ma_values):
+            return ma_values[-(1 + n_candles_ago)]
+        return None
+
+    def _get_rsi_value(self, coin, timeframe, period, n_candles_ago=0):
+        """Get RSI value from n candles ago (0 = latest, 1 = second-to-last)."""
+        closes = self._get_all_closes(coin, timeframe)
+        if not closes:
+            return None
+        rsi_values = self._compute_rsi(closes, period)
+        if not rsi_values:
+            return None
+        if n_candles_ago < len(rsi_values):
+            return rsi_values[-(1 + n_candles_ago)]
+        return None
+
+    def _format_change(self, current, past):
+        """Compute percentage change between two values."""
+        if past is None or past == 0 or current is None:
+            return None
+        return (current - past) / past * 100
+
+    def calculate_indicator(self, ind_def):
+        """
+        Calculate a single indicator based on its definition.
+        Returns a dict with value, changes, reference, and deviation.
+        """
+        ind_type = ind_def.get("type", "price")
+
+        if ind_type == "ratio":
+            return self._calc_ratio(ind_def)
+        elif ind_type == "price":
+            return self._calc_price(ind_def)
+        elif ind_type == "spread":
+            return self._calc_spread(ind_def)
+        elif ind_type == "diff_pct":
+            return self._calc_diff_pct(ind_def)
+        elif ind_type == "ma":
+            return self._calc_ma(ind_def)
+        elif ind_type == "rsi":
+            return self._calc_rsi(ind_def)
+        elif ind_type == "custom":
+            return self._calc_custom(ind_def)
+        else:
+            logging.warning(f"Unknown indicator type: {ind_type}")
+            return None
+
+    def _calc_ratio(self, ind_def):
+        """Calculate a ratio indicator (numerator / denominator)."""
+        num = ind_def["numerator"]
+        den = ind_def["denominator"]
+
+        num_now = self._get_latest_close(num)
+        den_now = self._get_latest_close(den)
+        if num_now is None or den_now is None or den_now == 0:
+            return None
+        current = num_now / den_now
+
+        changes = {}
+        for period in ind_def.get("changes", []):
+            num_past = self._get_close_n_candles_ago(num, period, n=1)
+            den_past = self._get_close_n_candles_ago(den, period, n=1)
+            if num_past is not None and den_past is not None and den_past != 0:
+                past = num_past / den_past
+                changes[period] = self._format_change(current, past)
+            else:
+                changes[period] = None
+
+        reference = None
+        deviation = None
+        if ind_def.get("show_deviation", False):
+            ratios = self._get_all_ratio(num, den, "1d")
+            if ratios:
+                min_points = ind_def.get("min_data_points", 100)
+                fallback_ref = ind_def.get("fallback_reference")
+                if len(ratios) < min_points and fallback_ref is not None:
+                    reference = fallback_ref
+                else:
+                    reference = sum(ratios) / len(ratios)
+                deviation = self._format_change(current, reference)
+
+        return {"value": current, "reference": reference, "changes": changes, "deviation": deviation}
+
+    def _calc_price(self, ind_def):
+        """Calculate a single price indicator."""
+        coin = ind_def["coin"]
+
+        current = self._get_latest_close(coin)
+        if current is None:
+            return None
+
+        changes = {}
+        for period in ind_def.get("changes", []):
+            past = self._get_close_n_candles_ago(coin, period, n=1)
+            changes[period] = self._format_change(current, past)
+
+        reference = None
+        deviation = None
+        if ind_def.get("show_deviation", False):
+            closes = self._get_all_closes(coin, "1d")
+            if closes:
+                min_points = ind_def.get("min_data_points", 100)
+                fallback_ref = ind_def.get("fallback_reference")
+                if len(closes) < min_points and fallback_ref is not None:
+                    reference = fallback_ref
+                else:
+                    reference = sum(closes) / len(closes)
+                deviation = self._format_change(current, reference)
+
+        return {"value": current, "reference": reference, "changes": changes, "deviation": deviation}
+
+    def _calc_spread(self, ind_def):
+        """Calculate a spread indicator (numerator - denominator)."""
+        num = ind_def["numerator"]
+        den = ind_def["denominator"]
+
+        num_now = self._get_latest_close(num)
+        den_now = self._get_latest_close(den)
+        if num_now is None or den_now is None:
+            return None
+        current = num_now - den_now
+
+        changes = {}
+        for period in ind_def.get("changes", []):
+            num_past = self._get_close_n_candles_ago(num, period, n=1)
+            den_past = self._get_close_n_candles_ago(den, period, n=1)
+            if num_past is not None and den_past is not None:
+                past = num_past - den_past
+                changes[period] = self._format_change(current, past)
+            else:
+                changes[period] = None
+
+        reference = None
+        deviation = None
+        if ind_def.get("show_deviation", False):
+            spreads = self._get_all_spread(num, den, "1d")
+            if spreads:
+                min_points = ind_def.get("min_data_points", 100)
+                fallback_ref = ind_def.get("fallback_reference")
+                if len(spreads) < min_points and fallback_ref is not None:
+                    reference = fallback_ref
+                else:
+                    reference = sum(spreads) / len(spreads)
+                deviation = self._format_change(current, reference)
+
+        return {"value": current, "reference": reference, "changes": changes, "deviation": deviation}
+
+    def _calc_diff_pct(self, ind_def):
+        """Calculate a percentage difference indicator ((num-den)/den*100)."""
+        num = ind_def["numerator"]
+        den = ind_def["denominator"]
+
+        num_now = self._get_latest_close(num)
+        den_now = self._get_latest_close(den)
+        if num_now is None or den_now is None or den_now == 0:
+            return None
+        current = (num_now - den_now) / den_now * 100
+
+        changes = {}
+        for period in ind_def.get("changes", []):
+            num_past = self._get_close_n_candles_ago(num, period, n=1)
+            den_past = self._get_close_n_candles_ago(den, period, n=1)
+            if num_past is not None and den_past is not None and den_past != 0:
+                past = (num_past - den_past) / den_past * 100
+                changes[period] = self._format_change(current, past)
+            else:
+                changes[period] = None
+
+        reference = None
+        deviation = None
+        if ind_def.get("show_deviation", False):
+            diffs = self._get_all_diff_pct(num, den, "1d")
+            if diffs:
+                min_points = ind_def.get("min_data_points", 100)
+                fallback_ref = ind_def.get("fallback_reference")
+                if len(diffs) < min_points and fallback_ref is not None:
+                    reference = fallback_ref
+                else:
+                    reference = sum(diffs) / len(diffs)
+                deviation = self._format_change(current, reference)
+
+        return {"value": current, "reference": reference, "changes": changes, "deviation": deviation}
+
+    def _calc_ma(self, ind_def):
+        """Calculate a moving average indicator."""
+        coin = ind_def["coin"]
+        timeframe = ind_def.get("timeframe", "1h")
+        period = ind_def.get("period", 20)
+
+        current = self._get_ma_value(coin, timeframe, period, n_candles_ago=0)
+        if current is None:
+            return None
+
+        changes = {}
+        for period_label in ind_def.get("changes", []):
+            if period_label == "1h":
+                past = self._get_ma_value(coin, "1h", period, n_candles_ago=1)
+            elif period_label == "1d":
+                past = self._get_ma_value(coin, "1d", period, n_candles_ago=1)
+            else:
+                past = self._get_ma_value(coin, period_label, period, n_candles_ago=1)
+            changes[period_label] = self._format_change(current, past)
+
+        reference = None
+        deviation = None
+        if ind_def.get("show_deviation", False):
+            live_price = self._get_latest_close(coin)
+            if live_price is not None and current != 0:
+                reference = current
+                deviation = (live_price - current) / current * 100
+
+        return {"value": current, "reference": reference, "changes": changes, "deviation": deviation}
+
+    def _calc_rsi(self, ind_def):
+        """Calculate an RSI indicator."""
+        coin = ind_def["coin"]
+        timeframe = ind_def.get("timeframe", "1h")
+        period = ind_def.get("period", 14)
+
+        current = self._get_rsi_value(coin, timeframe, period, n_candles_ago=0)
+        if current is None:
+            return None
+
+        changes = {}
+        for period_label in ind_def.get("changes", []):
+            if period_label == "1h":
+                past = self._get_rsi_value(coin, "1h", period, n_candles_ago=1)
+            elif period_label == "1d":
+                past = self._get_rsi_value(coin, "1d", period, n_candles_ago=1)
+            else:
+                past = self._get_rsi_value(coin, period_label, period, n_candles_ago=1)
+            if past is not None:
+                changes[period_label] = current - past
+            else:
+                changes[period_label] = None
+
+        reference = 50.0
+        deviation = None
+        if ind_def.get("show_deviation", False):
+            deviation = current - 50.0
+
+        return {"value": current, "reference": reference, "changes": changes, "deviation": deviation}
+
+    def _calc_custom(self, ind_def):
+        """Calculate a custom indicator by calling a user-defined function."""
+        module_path = ind_def.get("module")
+        function_name = ind_def.get("function")
+        args = ind_def.get("args", {})
+
+        if not module_path or not function_name:
+            logging.error(f"Custom indicator missing 'module' or 'function': {ind_def}")
+            return None
+
+        try:
+            module = importlib.import_module(module_path)
+            func = getattr(module, function_name)
+        except (ImportError, AttributeError) as e:
+            logging.error(f"Failed to load custom indicator {module_path}.{function_name}: {e}")
+            return None
+
+        try:
+            result = func(self.db_path, **args)
+            if not isinstance(result, dict):
+                logging.error(f"Custom indicator {function_name} must return a dict, got {type(result)}")
+                return None
+            return result
+        except Exception as e:
+            logging.error(f"Custom indicator {function_name} raised an error: {e}", exc_info=True)
+            return None
+
+    def calculate_all(self):
+        """Calculate all indicators defined in the config file."""
+        results = {}
+        for name, ind_def in self.config.items():
+            result = self.calculate_indicator(ind_def)
+            if result:
+                results[name] = {
+                    "display_name": ind_def.get("display_name", name),
+                    **result
+                }
+            else:
+                results[name] = {
+                    "display_name": ind_def.get("display_name", name),
+                    "value": None,
+                    "reference": None,
+                    "changes": {},
+                    "deviation": None
+                }
+        return results

indicators_fetcher.py

@@ -0,0 +1,86 @@
+"""
+Indicators Data Fetcher
+
+A standalone process that runs in a loop to compute financial indicators
+(ratios, prices, MAs, RSI, custom) from SQLite candle data and save
+the results to a JSON status file for the main dashboard to display.
+
+Follows the same pattern as dashboard_data_fetcher.py.
+"""
+
+import logging
+import os
+import sys
+import json
+import time
+import argparse
+from datetime import datetime, timezone
+
+from logging_utils import setup_logging
+from indicators import IndicatorCalculator
+
+
+class IndicatorsFetcher:
+    """
+    Periodically computes all configured indicators and saves them to a JSON file.
+    """
+
+    def __init__(self, log_level: str):
+        setup_logging(log_level, 'IndicatorsFetcher')
+
+        project_root = os.path.dirname(os.path.abspath(__file__))
+        self.db_path = os.environ.get("PG_CONN_STR", "postgresql://hyper:hyper@localhost:5432/hyper")
+        self.config_path = os.path.join(project_root, "_data", "indicators.json")
+        self.status_file_path = os.path.join(project_root, "_logs", "indicators_status.json")
+
+        self.calculator = IndicatorCalculator(
+            config_path=self.config_path,
+            db_path=self.db_path
+        )
+
+        logging.info(f"Indicators Fetcher initialized. DB: {self.db_path}, Config: {self.config_path}")
+
+    def fetch_and_save_indicators(self):
+        """Compute all indicators and save to JSON status file."""
+        try:
+            results = self.calculator.calculate_all()
+
+            status = {
+                "last_updated_utc": datetime.now(timezone.utc).isoformat(),
+                "indicators": results
+            }
+
+            logs_dir = os.path.dirname(self.status_file_path)
+            os.makedirs(logs_dir, exist_ok=True)
+
+            temp_file_path = self.status_file_path + ".tmp"
+            with open(temp_file_path, 'w', encoding='utf-8') as f:
+                json.dump(status, f, indent=4, default=str)
+            os.replace(temp_file_path, self.status_file_path)
+
+            logging.debug(f"Successfully updated indicators status file with {len(results)} indicators.")
+
+        except Exception as e:
+            logging.error(f"Failed to fetch or save indicators: {e}", exc_info=True)
+
+    def run(self):
+        """Main loop to periodically compute and save indicators."""
+        logging.info("Starting Indicators Fetcher loop (update interval: 30s)")
+        while True:
+            try:
+                self.fetch_and_save_indicators()
+            except Exception as e:
+                logging.error(f"Indicators Fetcher loop error: {e}", exc_info=True)
+            time.sleep(30)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Run the Indicators Data Fetcher.")
+    parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug'])
+    args = parser.parse_args()
+
+    fetcher = IndicatorsFetcher(log_level=args.log_level)
+    try:
+        fetcher.run()
+    except KeyboardInterrupt:
+        logging.info("Indicators Data Fetcher stopped.")

live_candle_fetcher.py

@@ -7,7 +7,7 @@ import time
 from datetime import datetime, timezone
 from hyperliquid.info import Info
 from hyperliquid.utils import constants
-import sqlite3
+import db
 from queue import Queue
 from threading import Thread
 
@@ -22,7 +22,7 @@ class LiveCandleFetcher:
 
     def __init__(self, log_level: str, coins: list):
         setup_logging(log_level, 'LiveCandleFetcher')
-        self.db_path = os.path.join("_data", "market_data.db")
+        self.db_path = os.environ.get("PG_CONN_STR", "postgresql://hyper:hyper@localhost:5432/hyper")
         self.coins_to_watch = set(coins)
         if not self.coins_to_watch:
             logging.error("No coins provided to watch. Exiting.")
@@ -34,66 +34,16 @@ class LiveCandleFetcher:
 
     def _ensure_tables_exist(self):
         """
-        Ensures that all necessary tables are created with the correct schema and PRIMARY KEY.
+        Ensures that all necessary tables are created with the correct schema.
-        If a table exists with an incorrect schema, it attempts to migrate the data.
+        Uses db.create_candle_table() which is idempotent (CREATE TABLE IF NOT EXISTS).
         """
-        with sqlite3.connect(self.db_path) as conn:
+        conn = db.get_connection()
         for coin in self.coins_to_watch:
-                table_name = f"{coin}_1m"
+            table_name = db.sanitize_table_name(coin, "1m")
-                cursor = conn.cursor()
+            db.create_candle_table(conn, table_name)
-                cursor.execute(f"PRAGMA table_info('{table_name}')")
+        conn.close()
-                columns = cursor.fetchall()
-                
-                if columns:
-                    pk_found = any(col[1] == 'timestamp_ms' and col[5] == 1 for col in columns)
-                    if not pk_found:
-                        logging.warning(f"Schema migration needed for table '{table_name}': 'timestamp_ms' is not the PRIMARY KEY.")
-                        logging.warning("Attempting to automatically rebuild the table...")
-                        try:
-                            # 1. Rename old table
-                            conn.execute(f'ALTER TABLE "{table_name}" RENAME TO "{table_name}_old"')
-                            logging.info(f"  -> Renamed existing table to '{table_name}_old'.")
-                            
-                            # 2. Create new table with correct schema
-                            self._create_candle_table(conn, table_name)
-                            logging.info(f"  -> Created new '{table_name}' table with correct schema.")
-                            
-                            # 3. Copy unique data from old table to new table
-                            conn.execute(f'''
-                                INSERT OR IGNORE INTO "{table_name}" (datetime_utc, timestamp_ms, open, high, low, close, volume, number_of_trades)
-                                SELECT datetime_utc, timestamp_ms, open, high, low, close, volume, number_of_trades
-                                FROM "{table_name}_old"
-                            ''')
-                            conn.commit()
-                            logging.info("  -> Copied data to new table.")
-
-                            # 4. Drop the old table
-                            conn.execute(f'DROP TABLE "{table_name}_old"')
-                            logging.info(f"  -> Removed old table. Migration for '{table_name}' complete.")
-                        except Exception as e:
-                            logging.error(f"FATAL: Automatic schema migration for '{table_name}' failed: {e}")
-                            logging.error("Please delete the database file '_data/market_data.db' manually and restart.")
-                            sys.exit(1)
-                else:
-                    # If table does not exist, create it
-                    self._create_candle_table(conn, table_name)
         logging.info("Database tables verified.")
 
-    def _create_candle_table(self, conn, table_name: str):
-        """Creates a new candle table with the correct schema."""
-        conn.execute(f'''
-            CREATE TABLE "{table_name}" (
-                datetime_utc TEXT,
-                timestamp_ms INTEGER PRIMARY KEY,
-                open REAL,
-                high REAL,
-                low REAL,
-                close REAL,
-                volume REAL,
-                number_of_trades INTEGER
-            )
-        ''')
-
     def on_message(self, message):
         """
         Callback function to process incoming candle messages. This is the "Producer".
@@ -112,6 +62,7 @@ class LiveCandleFetcher:
         This is the "Consumer" thread. It runs forever, pulling candles from the
         queue and writing them to the database, ensuring all writes are serial.
         """
+        conn = db.get_connection()
         while True:
             try:
                 candle = self.candle_queue.get()
@@ -122,7 +73,7 @@ class LiveCandleFetcher:
                 if not coin:
                     continue
 
-                table_name = f"{coin}_1m"
+                table_name = db.sanitize_table_name(coin, "1m")
                 record = (
                     datetime.fromtimestamp(candle['t'] / 1000, tz=timezone.utc).strftime('%Y-%m-%d %H:%M:%S'),
                     candle['t'],
@@ -130,24 +81,21 @@ class LiveCandleFetcher:
                     candle.get('v'), candle.get('n')
                 )
                 
-                with sqlite3.connect(self.db_path) as conn:
+                db.upsert_candles(conn, table_name, [record])
-                    conn.execute(f'''
-                        INSERT OR REPLACE INTO "{table_name}" (datetime_utc, timestamp_ms, open, high, low, close, volume, number_of_trades)
-                        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
-                    ''', record)
-                    conn.commit()
                 logging.debug(f"Upserted candle for {coin} at {record[0]}")
 
             except Exception as e:
                 logging.error(f"Error in database writer thread: {e}")
+        conn.close()
 
     def _get_last_timestamp_from_db(self, coin: str) -> int:
         """Gets the most recent millisecond timestamp from a coin's 1m table."""
-        table_name = f"{coin}_1m"
+        table_name = db.sanitize_table_name(coin, "1m")
         try:
-            with sqlite3.connect(self.db_path) as conn:
+            conn = db.get_connection()
-                result = conn.execute(f'SELECT MAX(timestamp_ms) FROM "{table_name}"').fetchone()
+            result = db.get_last_timestamp(conn, table_name)
-                return int(result[0]) if result and result[0] is not None else None
+            conn.close()
+            return result
         except Exception as e:
             logging.error(f"Could not read last timestamp from table '{table_name}': {e}")
             return None

main_app.py

@@ -8,47 +8,42 @@ import multiprocessing
 import schedule
 import sqlite3
 import pandas as pd
-from datetime import datetime, timezone
+from datetime import datetime
 import importlib
+from dotenv import load_dotenv
+load_dotenv()
 # --- REMOVED: import signal ---
 # --- REMOVED: from queue import Empty ---
 
 from logging_utils import setup_logging
-# --- Using the new high-performance WebSocket utility for live prices ---
 from live_market_utils import start_live_feed
-# --- Import the base class for type hinting (optional but good practice) ---
 from strategies.base_strategy import BaseStrategy
+from dashboard import DashboardRenderer
+from rich.live import Live
+from hyperliquid.info import Info
+from hyperliquid.utils import constants
 
 # --- Configuration ---
-WATCHED_COINS = ["BTC", "ETH", "SOL", "BNB", "HYPE", "ASTER", "ZEC", "PUMP", "SUI", "xyz:BRENTOIL", "xyz:CL"]
+WATCHED_COINS = ["BTC", "ETH", "SOL", "BNB", "HYPE", "SUI", "xyz:BRENTOIL", "xyz:CL", "xyz:GOLD", "xyz:SILVER", "mkts:USTECH", "xyz:XYZ100"]
 # Display name mapping for dashboard (internal symbol -> display name)
 COIN_DISPLAY_NAMES = {
     "xyz:BRENTOIL": "BRENT",
-    "xyz:CL": "WTI"
+    "xyz:CL": "WTI",
+    "xyz:GOLD": "GOLD",
+    "xyz:SILVER": "SILVER",
+    "mkts:USTECH": "USTECH",
+    "xyz:XYZ100": "XYZ100"
 }
 LIVE_CANDLE_FETCHER_SCRIPT = "live_candle_fetcher.py"
 RESAMPLER_SCRIPT = "resampler.py" 
 # --- REMOVED: Market Cap Fetcher ---
 # --- REMOVED: trade_executor.py is no longer a script ---
 DASHBOARD_DATA_FETCHER_SCRIPT = "dashboard_data_fetcher.py"
+INDICATORS_FETCHER_SCRIPT = "indicators_fetcher.py"
 STRATEGY_CONFIG_FILE = os.path.join("_data", "strategies.json")
 DB_PATH = os.path.join("_data", "market_data.db")
 # --- REMOVED: Market Cap File ---
 LOGS_DIR = "_logs"
-TRADE_EXECUTOR_STATUS_FILE = os.path.join(LOGS_DIR, "trade_executor_status.json")
-
-
-def format_market_cap(mc_value):
-    """Formats a large number into a human-readable market cap string."""
-    if not isinstance(mc_value, (int, float)) or mc_value == 0:
-        return "N/A"
-    if mc_value >= 1_000_000_000_000:
-        return f"${mc_value / 1_000_000_000_000:.2f}T"
-    if mc_value >= 1_000_000_000:
-        return f"${mc_value / 1_000_000_000:.2f}B"
-    if mc_value >= 1_000_000:
-        return f"${mc_value / 1_000_000:.2f}M"
-    return f"${mc_value:,.2f}"
 
 
 def run_live_candle_fetcher():
@@ -348,17 +343,60 @@ def run_dashboard_data_fetcher():
             time.sleep(10)
 
 
+def run_indicators_fetcher():
+    """Target function to run the indicators_fetcher.py script."""
+
+    # --- GRACEFUL SHUTDOWN HANDLER ---
+    import signal
+
+    def handle_shutdown_signal(signum, frame):
+        try:
+            logging.info(f"Shutdown signal ({signum}) received. Initiating graceful exit...")
+        except NameError:
+            print(f"[IndicatorsFetcher] Shutdown signal ({signum}) received. Initiating graceful exit...")
+        raise KeyboardInterrupt
+
+    signal.signal(signal.SIGTERM, handle_shutdown_signal)
+    # --- END GRACEFUL SHUTDOWN HANDLER ---
+
+    log_file = os.path.join(LOGS_DIR, "indicators_fetcher.log")
+    while True:
+        try:
+            with open(log_file, 'a') as f:
+                f.write(f"\n--- Starting Indicators Fetcher at {datetime.now()} ---\n")
+                subprocess.run([sys.executable, INDICATORS_FETCHER_SCRIPT, "--log-level", "normal"], check=True, stdout=f, stderr=subprocess.STDOUT)
+        except KeyboardInterrupt:
+            logging.info("Indicators Fetcher stopping.")
+            break
+        except (subprocess.CalledProcessError, Exception) as e:
+            with open(log_file, 'a') as f:
+                f.write(f"\n--- PROCESS ERROR at {datetime.now()} ---\n")
+                f.write(f"Indicators Fetcher failed: {e}. Restarting...\n")
+            time.sleep(10)
+
+
 class MainApp:
     def __init__(self, coins_to_watch: list, processes: dict, strategy_configs: dict, shared_prices: dict):
         self.watched_coins = coins_to_watch
         self.shared_prices = shared_prices
         self.prices = {}
-        # --- REMOVED: self.market_caps ---
-        self.open_positions = {}
         self.background_processes = processes
         self.process_status = {}
         self.strategy_configs = strategy_configs
         self.strategy_statuses = {}
+        self.indicators_status = {}
+        self.account_data = None
+        self.wallet_address = os.environ.get("MAIN_WALLET_ADDRESS")
+        if self.wallet_address:
+            self.info_client = Info(constants.MAINNET_API_URL, skip_ws=True)
+        else:
+            self.info_client = None
+        self.renderer = DashboardRenderer(table_visibility={
+            "market": True,
+            "strategies": False,
+            "indicators": True,
+            "balances": True,
+        })
 
     def read_prices(self):
         """Reads the latest prices directly from the shared memory dictionary."""
@@ -386,189 +424,77 @@ class MainApp:
                     enabled_statuses[name] = {"current_signal": "Initializing..."}
         self.strategy_statuses = enabled_statuses
 
-    def read_executor_status(self):
+    def read_indicators_status(self):
-        """Reads the live status file from the trade executor."""
+        """Reads the indicators status JSON file."""
-        if os.path.exists(TRADE_EXECUTOR_STATUS_FILE):
+        status_file = os.path.join(LOGS_DIR, "indicators_status.json")
+        if os.path.exists(status_file):
             try:
-                with open(TRADE_EXECUTOR_STATUS_FILE, 'r', encoding='utf-8') as f:
+                with open(status_file, 'r', encoding='utf-8') as f:
-                    # --- FIX: Read the 'open_positions' key from the file ---
+                    self.indicators_status = json.load(f)
-                    status_data = json.load(f)
-                    self.open_positions = status_data.get('open_positions', {})
             except (IOError, json.JSONDecodeError):
-                logging.debug("Could not read trade executor status file.")
+                self.indicators_status = {}
         else:
-            self.open_positions = {}
+            self.indicators_status = {}
+
+    def read_account_data(self):
+        """Fetches account balances and positions from Hyperliquid API."""
+        if not self.wallet_address or not self.info_client:
+            self.account_data = None
+            return
+        try:
+            perp_state = self.info_client.user_state(self.wallet_address)
+            spot_state = self.info_client.spot_user_state(self.wallet_address)
+
+            margin_summary = perp_state.get('marginSummary', {})
+            account_value = float(margin_summary.get('accountValue', 0))
+            margin_used = float(margin_summary.get('totalMarginUsed', 0))
+            utilization = (margin_used / account_value) * 100 if account_value > 0 else 0
+
+            spot_balances = spot_state.get('balances', [])
+            positions = perp_state.get('assetPositions', [])
+
+            self.account_data = {
+                'account_value': account_value,
+                'margin_used': margin_used,
+                'utilization': utilization,
+                'spot_balances': spot_balances,
+                'positions': positions,
+            }
+
+        except Exception as e:
+            logging.error(f"Could not fetch account data: {e}")
+            self.account_data = None
 
     def check_process_status(self):
         """Checks if the background processes are still running."""
         for name, process in self.background_processes.items():
             self.process_status[name] = "Running" if process.is_alive() else "STOPPED"
 
-    def _format_price(self, price_val, width=10):
+    def toggle_table(self, table_name, enabled=None):
-        """Helper function to format prices for the dashboard."""
+        """Toggle a dashboard table's visibility at runtime."""
-        try:
+        return self.renderer.toggle_table(table_name, enabled)
-            price_float = float(price_val)
-            if price_float < 1:
-                price_str = f"{price_float:>{width}.6f}"
-            elif price_float < 100:
-                price_str = f"{price_float:>{width}.4f}"
-            else:
-                price_str = f"{price_float:>{width}.2f}"
-        except (ValueError, TypeError):
-            price_str = f"{'Loading...':>{width}}"
-        return price_str
 
     def display_dashboard(self):
-        """Displays a formatted dashboard with side-by-side tables."""
+        """Build and return the rich dashboard layout."""
-        print("\x1b[H\x1b[J", end="") # Clear screen
+        return self.renderer.build_layout(
-        
+            self.watched_coins,
-        left_table_lines = ["--- Market Dashboard ---"]
+            self.prices,
-        # --- MODIFIED: Adjusted width for new columns ---
+            COIN_DISPLAY_NAMES,
-        left_table_width = 65
+            self.strategy_statuses,
-        left_table_lines.append("-" * left_table_width)
+            self.strategy_configs,
-        # --- MODIFIED: Replaced Market Cap with Gap ---
+            self.indicators_status,
-        left_table_lines.append(f"{'#':<2} | {'Coin':^6} | {'Best Bid':>10} | {'Live Price':>10} | {'Best Ask':>10} | {'Gap':>10} |")
+            self.account_data
-        left_table_lines.append("-" * left_table_width)
+        )
-        for i, coin in enumerate(self.watched_coins, 1):
-            # Use display name for dashboard, but keep internal symbol for price lookup
-            display_name = COIN_DISPLAY_NAMES.get(coin, coin)
-            
-            # --- MODIFIED: Fetch all three price types ---
-            mid_price = self.prices.get(coin, "Loading...")
-            bid_price = self.prices.get(f"{coin}_bid", "Loading...")
-            ask_price = self.prices.get(f"{coin}_ask", "Loading...")
-
-            # --- MODIFIED: Use the new formatting helper ---
-            formatted_mid = self._format_price(mid_price)
-            formatted_bid = self._format_price(bid_price)
-            formatted_ask = self._format_price(ask_price)
-
-            # --- MODIFIED: Calculate gap ---
-            gap_str = f"{'Loading...':>10}"
-            try:
-                # Calculate the spread
-                gap_val = float(ask_price) - float(bid_price)
-                # Format gap with high precision, similar to price
-                if gap_val < 1:
-                    gap_str = f"{gap_val:>{10}.6f}"
-                else:
-                    gap_str = f"{gap_val:>{10}.4f}"
-            except (ValueError, TypeError):
-                pass # Keep 'Loading...'
-
-            # --- REMOVED: Market Cap logic ---
-            
-            # --- MODIFIED: Print all price columns including gap ---
-            left_table_lines.append(f"{i:<2} | {display_name:^6} | {formatted_bid} | {formatted_mid} | {formatted_ask} | {gap_str} |")
-        left_table_lines.append("-" * left_table_width)
-
-        right_table_lines = ["--- Strategy Status ---"]
-        # --- FIX: Adjusted table width after removing parameters ---
-        right_table_width = 105
-        right_table_lines.append("-" * right_table_width)
-        # --- FIX: Removed 'Parameters' from header ---
-        right_table_lines.append(f"{'#':^2} | {'Strategy Name':<25} | {'Coin':^6} | {'Signal':^8} | {'Signal Price':>12} | {'Last Change':>17} | {'TF':^5} | {'Size':^8} |")
-        right_table_lines.append("-" * right_table_width)
-        for i, (name, status) in enumerate(self.strategy_statuses.items(), 1):
-            signal = status.get('current_signal', 'N/A')
-            price = status.get('signal_price')
-            price_display = f"{price:.4f}" if isinstance(price, (int, float)) else "-"
-            last_change = status.get('last_signal_change_utc')
-            last_change_display = 'Never'
-            if last_change:
-                dt_utc = datetime.fromisoformat(last_change.replace('Z', '+00:00')).replace(tzinfo=timezone.utc)
-                dt_local = dt_utc.astimezone(None)
-                last_change_display = dt_local.strftime('%Y-%m-%d %H:%M')
-            
-            config_params = self.strategy_configs.get(name, {}).get('parameters', {})
-            
-            # --- FIX: Read coin/size from status file first, fallback to config ---
-            coin = status.get('coin', config_params.get('coin', 'N/A'))
-            
-            # --- FIX: Handle nested 'coins_to_copy' logic for size ---
-            # --- MODIFIED: Read 'size' from status first, then config, then 'Multi' ---
-            size = status.get('size')
-            if not size:
-                if 'coins_to_copy' in config_params:
-                    size = 'Multi'
-                else:
-                    size = config_params.get('size', 'N/A')
-            
-            timeframe = config_params.get('timeframe', 'N/A')
-            
-            # --- FIX: Removed parameter string logic ---
-            
-            # --- FIX: Removed 'params_str' from the formatted line ---
-                
-            size_display = f"{size:>8}"
-            if isinstance(size, (int, float)):
-                # --- MODIFIED: More flexible size formatting ---
-                if size < 0.0001:
-                    size_display = f"{size:>8.6f}"
-                elif size < 1:
-                    size_display = f"{size:>8.4f}"
-                else:
-                    size_display = f"{size:>8.2f}"
-            # --- END NEW LOGIC ---
-            
-            right_table_lines.append(f"{i:^2} | {name:<25} | {coin:^6} | {signal:^8} | {price_display:>12} | {last_change_display:>17} | {timeframe:^5} | {size_display} |")
-        right_table_lines.append("-" * right_table_width)
-        
-        output_lines = []
-        max_rows = max(len(left_table_lines), len(right_table_lines))
-        separator = "   "
-        indent = " " * 10
-        for i in range(max_rows):
-            left_part = left_table_lines[i] if i < len(left_table_lines) else " " * left_table_width
-            right_part = indent + right_table_lines[i] if i < len(right_table_lines) else ""
-            output_lines.append(f"{left_part}{separator}{right_part}")
-        
-        output_lines.append("\n--- Open Positions ---")
-        pos_table_width = 100
-        output_lines.append("-" * pos_table_width)
-        output_lines.append(f"{'Account':<10} | {'Coin':<6} | {'Size':>15} | {'Entry Price':>12} | {'Mark Price':>12} | {'PNL':>15} | {'Leverage':>10} |")
-        output_lines.append("-" * pos_table_width)
-
-        # --- FIX: Correctly read and display open positions ---
-        if not self.open_positions:
-            output_lines.append(f"{'No open positions.':^{pos_table_width}}")
-        else:
-            for account, positions in self.open_positions.items():
-                if not positions:
-                    continue
-                for coin, pos in positions.items():
-                    try:
-                        size_f = float(pos.get('size', 0))
-                        entry_f = float(pos.get('entry_price', 0))
-                        mark_f = float(self.prices.get(coin, 0))
-                        pnl_f = (mark_f - entry_f) * size_f if size_f > 0 else (entry_f - mark_f) * abs(size_f)
-                        lev = pos.get('leverage', 1)
-                        
-                        size_str = f"{size_f:>{15}.5f}"
-                        entry_str = f"{entry_f:>{12}.2f}"
-                        mark_str = f"{mark_f:>{12}.2f}"
-                        pnl_str = f"{pnl_f:>{15}.2f}"
-                        lev_str = f"{lev}x"
-                        
-                        output_lines.append(f"{account:<10} | {coin:<6} | {size_str} | {entry_str} | {mark_str} | {pnl_str} | {lev_str:>10} |")
-                    except (ValueError, TypeError):
-                         output_lines.append(f"{account:<10} | {coin:<6} | {'Error parsing data...':^{pos_table_width-20}} |")
-
-        output_lines.append("-" * pos_table_width)
-            
-        final_output = "\n".join(output_lines)
-        print(final_output)
-        sys.stdout.flush()
 
     def run(self):
         """Main loop to read data, display dashboard, and check processes."""
+        with Live(self.display_dashboard(), refresh_per_second=2, console=self.renderer.console) as live:
             while True:
                 self.read_prices()
-            # --- REMOVED: self.read_market_caps() ---
                 self.read_strategy_statuses()
-            self.read_executor_status()
+                self.read_indicators_status()
-            # --- REMOVED: self.check_process_status() ---
+                self.read_account_data()
-            self.display_dashboard()
+                live.update(self.display_dashboard())
                 time.sleep(0.5)
 
 if __name__ == "__main__":
@@ -613,6 +539,7 @@ if __name__ == "__main__":
         processes["Resampler"] = multiprocessing.Process(target=resampler_scheduler, args=(list(required_timeframes),), daemon=True)
         # --- REMOVED: Market Cap Fetcher Process ---
         processes["Dashboard Data"] = multiprocessing.Process(target=run_dashboard_data_fetcher, daemon=True)
+        processes["Indicators"] = multiprocessing.Process(target=run_indicators_fetcher, daemon=True)
 
         processes["Position Manager"] = multiprocessing.Process(
             target=run_position_manager, 

migrate_sqlite_to_pg.py

@@ -0,0 +1,105 @@
+#!/usr/bin/env python3
+"""
+SQLite to PostgreSQL Migration Script
+
+Reads candle data from a SQLite database and writes it to PostgreSQL.
+This is a one-time migration tool used to transfer existing historical
+data from the old SQLite database to the new PostgreSQL database.
+
+Usage:
+    python migrate_sqlite_to_pg.py --sqlite-path _data/market_data.db --log-level normal
+
+The script:
+1. Connects to both SQLite (source) and PostgreSQL (destination)
+2. Enumerates all candle tables (skipping legacy tables like market_cap)
+3. For each table, reads data from SQLite and upserts to PostgreSQL
+4. Handles table name sanitization (colons → underscores)
+"""
+
+import argparse
+import logging
+import os
+import sqlite3
+import sys
+
+import pandas as pd
+
+from logging_utils import setup_logging
+from db import get_connection, sanitize_table_name, upsert_candles, create_candle_table
+
+TIMEFRAMES = [
+    '1m', '3m', '5m', '15m', '30m', '37m', '148m',
+    '1h', '2h', '4h', '8h', '12h', '1d', '3d', '1w', '1month'
+]
+
+
+def get_candle_tables(sqlite_conn):
+    """Get all candle table names from SQLite (excluding legacy tables)."""
+    cursor = sqlite_conn.cursor()
+    cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
+    all_tables = [row[0] for row in cursor.fetchall()]
+
+    candle_tables = []
+    for table in all_tables:
+        for tf in TIMEFRAMES:
+            if table.endswith(f'_{tf}'):
+                candle_tables.append(table)
+                break
+
+    return candle_tables
+
+
+def parse_table_name(table_name):
+    """Parse a table name into (coin, timeframe)."""
+    for tf in TIMEFRAMES:
+        suffix = f'_{tf}'
+        if table_name.endswith(suffix):
+            coin = table_name[:-len(suffix)]
+            return coin, tf
+    return table_name, '1m'
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Migrate data from SQLite to PostgreSQL.")
+    parser.add_argument("--sqlite-path", default="_data/market_data.db",
+                        help="Path to the SQLite database file.")
+    parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug'])
+    args = parser.parse_args()
+
+    setup_logging(args.log_level, 'Migrator')
+
+    if not os.path.exists(args.sqlite_path):
+        logging.error(f"SQLite database not found at '{args.sqlite_path}'")
+        sys.exit(1)
+
+    sqlite_conn = sqlite3.connect(args.sqlite_path)
+    pg_conn = get_connection()
+
+    tables = get_candle_tables(sqlite_conn)
+    logging.info(f"Found {len(tables)} candle tables to migrate")
+
+    for table_name in tables:
+        coin, timeframe = parse_table_name(table_name)
+        pg_table = sanitize_table_name(coin, timeframe)
+
+        logging.info(f"Migrating {table_name} -> {pg_table}")
+
+        df = pd.read_sql(f'SELECT * FROM "{table_name}"', sqlite_conn)
+        if df.empty:
+            logging.warning(f"Table {table_name} is empty, skipping")
+            continue
+
+        create_candle_table(pg_conn, pg_table)
+
+        records = list(df.itertuples(index=False, name=None))
+        upsert_candles(pg_conn, pg_table, records)
+
+        logging.info(f"Migrated {len(records)} rows to {pg_table}")
+
+    sqlite_conn.close()
+    pg_conn.close()
+    logging.info("Migration complete!")
+
+
+if __name__ == "__main__":
+    main()

position_monitor.py

@@ -77,14 +77,13 @@ class PositionMonitor:
             output_lines.append("\n--- Perpetuals Account Summary ---")
             output_lines.append(f"  Account Value: ${account_value:,.2f}  |  Margin Used: ${margin_used:,.2f}  |  Utilization: {utilization:.2f}%")
             
-            # --- 2. Spot Balances Summary ---
+            # --- 2. Spot Balances Table ---
             output_lines.append("\n--- Spot Balances ---")
             spot_balances = spot_state.get('balances', [])
             if not spot_balances:
                 output_lines.append("  No spot balances found.")
             else:
-                balances_str = ", ".join([f"{b.get('coin')}: {float(b.get('total', 0)):,.4f}" for b in spot_balances if float(b.get('total', 0)) > 0])
+                self.build_spot_balances_table(spot_balances, output_lines)
-                output_lines.append(f"  {balances_str}")
                 
             # --- 3. Open Positions Table ---
             output_lines.append("\n--- Open Perpetual Positions ---")
@@ -106,6 +105,23 @@ class PositionMonitor:
         self._lines_printed = len(output_lines)
         sys.stdout.flush()
 
+    def build_spot_balances_table(self, spot_balances: list, output_lines: list):
+        """Builds the text for the spot balances table."""
+        header = f"| {'Coin':<10} | {'Total':>18} |"
+        output_lines.append(header)
+        output_lines.append("-" * len(header))
+        
+        for balance in spot_balances:
+            coin = balance.get('coin', 'Unknown')
+            total = float(balance.get('total', 0))
+            
+            coin_str = f"{coin:<10}"
+            total_str = f"{total:>18,.4f}"
+            
+            output_lines.append(f"| {coin_str} | {total_str} |")
+        
+        output_lines.append("-" * len(header))
+
     def build_positions_table(self, positions: list, coin_to_strategy_map: dict, output_lines: list):
         """Builds the text for the positions summary table."""
         header = f"| {'Strategy':<25} | {'Coin':<6} | {'Side':<5} | {'Size':>15} | {'Entry Price':>12} | {'Mark Price':>12} | {'PNL':>15} | {'Leverage':>10} |"

postgres/postgresql.conf

@@ -0,0 +1,26 @@
+# PostgreSQL configuration tuned for Synology DS1513+ (4GB RAM)
+# Place this file at postgres/postgresql.conf and mount it into the container.
+
+# --- Memory ---
+shared_buffers = 128MB
+effective_cache_size = 512MB
+work_mem = 8MB
+maintenance_work_mem = 64MB
+
+# --- Connections ---
+max_connections = 10
+max_worker_processes = 2
+
+# --- WAL / Checkpointing ---
+wal_buffers = 4MB
+checkpoint_completion_target = 0.9
+max_wal_senders = 3
+
+# --- Network ---
+listen_addresses = '*'
+
+# --- Logging ---
+log_statement = 'none'
+log_duration = off
+log_min_duration_statement = 0
+log_line_prefix = '%t [%p]: [%l-1] user=%u,db=%d,app=%a,client=%h '

requirements.txt

@@ -39,6 +39,7 @@ pydantic_core==2.41.5
 python-dateutil==2.9.0.post0
 python-dotenv==1.2.1
 pytz==2025.2
+rich==13.9.4
 regex==2025.11.3
 requests==2.32.5
 rlp==4.1.0
@@ -52,3 +53,4 @@ urllib3==1.26.20
 websocket-client==1.9.0
 web3~=6.0.0 # This means >=6.0.0 and <7.0.0
 yarl==1.22.0
+psycopg2-binary==2.9.9

resampler.py

@@ -2,7 +2,7 @@ import argparse
 import logging
 import os
 import sys
-import sqlite3
+import db
 import pandas as pd
 import json
 from datetime import datetime, timezone, timedelta
@@ -19,7 +19,7 @@ class Resampler:
 
     def __init__(self, log_level: str, coins: list, timeframes: dict):
         setup_logging(log_level, 'Resampler')
-        self.db_path = os.path.join("_data", "market_data.db")
+        self.db_path = os.environ.get("PG_CONN_STR", "postgresql://hyper:hyper@localhost:5432/hyper")
         self.status_file_path = os.path.join("_data", "resampling_status.json")
         self.coins_to_process = coins
         self.timeframes = timeframes
@@ -37,59 +37,17 @@ class Resampler:
 
     def _ensure_tables_exist(self):
         """
-        Ensures all resampled tables exist with a PRIMARY KEY on timestamp_ms.
+        Ensures all resampled tables exist with the correct schema.
-        Attempts to migrate existing tables if the schema is incorrect.
+        Uses db.create_candle_table() which is idempotent.
         """
-        with sqlite3.connect(self.db_path) as conn:
+        conn = db.get_connection()
         for coin in self.coins_to_process:
             for tf_name in self.timeframes.keys():
-                    table_name = f"{coin}_{tf_name}"
+                table_name = db.sanitize_table_name(coin, tf_name)
-                    cursor = conn.cursor()
+                db.create_candle_table(conn, table_name)
-                    cursor.execute(f"PRAGMA table_info('{table_name}')")
+        conn.close()
-                    columns = cursor.fetchall()
-                    if columns:
-                        # --- FIX: Check for the correct PRIMARY KEY on timestamp_ms ---
-                        pk_found = any(col[1] == 'timestamp_ms' and col[5] == 1 for col in columns)
-                        if not pk_found:
-                            logging.warning(f"Schema migration needed for table '{table_name}'.")
-                            try:
-                                conn.execute(f'ALTER TABLE "{table_name}" RENAME TO "{table_name}_old"')
-                                self._create_resampled_table(conn, table_name)
-                                # Copy data, ensuring to create the timestamp_ms
-                                logging.info(f"  -> Migrating data for '{table_name}'...")
-                                old_df = pd.read_sql(f'SELECT * FROM "{table_name}_old"', conn, parse_dates=['datetime_utc'])
-                                if not old_df.empty:
-                                    old_df['timestamp_ms'] = (old_df['datetime_utc'].astype('int64') // 10**6)
-                                    # Keep only unique timestamps, preserving the last entry
-                                    old_df.drop_duplicates(subset=['timestamp_ms'], keep='last', inplace=True)
-                                    old_df.to_sql(table_name, conn, if_exists='append', index=False)
-                                logging.info(f"  -> Data migration complete.")
-                                conn.execute(f'DROP TABLE "{table_name}_old"')
-                                conn.commit()
-                                logging.info(f"Successfully migrated schema for '{table_name}'.")
-                            except Exception as e:
-                                logging.error(f"FATAL: Migration for '{table_name}' failed: {e}. Please delete 'market_data.db' and restart.")
-                                sys.exit(1)
-                    else:
-                        self._create_resampled_table(conn, table_name)
         logging.info("All resampled table schemas verified.")
 
-    def _create_resampled_table(self, conn, table_name):
-        """Creates a new resampled table with the correct schema."""
-        # --- FIX: Set PRIMARY KEY on timestamp_ms for performance and uniqueness ---
-        conn.execute(f'''
-            CREATE TABLE "{table_name}" (
-                datetime_utc TEXT,
-                timestamp_ms INTEGER PRIMARY KEY,
-                open REAL,
-                high REAL,
-                low REAL,
-                close REAL,
-                volume REAL,
-                number_of_trades INTEGER
-            )
-        ''')
-
     def _load_existing_status(self) -> dict:
         """Loads the existing status file if it exists, otherwise returns an empty dict."""
         if os.path.exists(self.status_file_path):
@@ -116,13 +74,8 @@ class Resampler:
             logging.warning("No timeframes to process after filtering. Exiting job.")
             return
 
-        if not os.path.exists(self.db_path):
+        conn = db.get_connection()
-            logging.error(f"Database file '{self.db_path}' not found.")
+        try:
-            return
-
-        with sqlite3.connect(self.db_path) as conn:
-            conn.execute("PRAGMA journal_mode=WAL;")
-            
             logging.debug(f"Processing {len(self.coins_to_process)} coins...")
 
             for coin in self.coins_to_process:
@@ -130,8 +83,8 @@ class Resampler:
                 
                 try:
                     for tf_name, tf_code in self.timeframes.items():
-                        target_table_name = f"{coin}_{tf_name}"
+                        target_table_name = db.sanitize_table_name(coin, tf_name)
-                        source_table_name = f"{coin}_1m"
+                        source_table_name = db.sanitize_table_name(coin, "1m")
                         logging.debug(f"  Updating {tf_name} table...")
                         
                         last_timestamp_ms = self._get_last_timestamp(conn, target_table_name)
@@ -139,7 +92,7 @@ class Resampler:
                         query = f'SELECT * FROM "{source_table_name}"'
                         params = ()
                         if last_timestamp_ms:
-                            query += ' WHERE timestamp_ms >= ?'
+                            query += ' WHERE timestamp_ms >= %s'
                             # Go back one interval to rebuild the last (potentially partial) candle
                             try:
                                 interval_delta_ms = pd.to_timedelta(tf_code).total_seconds() * 1000
@@ -170,12 +123,7 @@ class Resampler:
                                     row['volume'], row['number_of_trades']
                                 ))
                             
-                            cursor = conn.cursor()
+                            db.upsert_candles(conn, target_table_name, records_to_upsert)
-                            cursor.executemany(f'''
-                                INSERT OR REPLACE INTO "{target_table_name}" (datetime_utc, timestamp_ms, open, high, low, close, volume, number_of_trades)
-                                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
-                            ''', records_to_upsert)
-                            conn.commit()
 
                             logging.debug(f"    -> Upserted {len(resampled_df)} candles into '{target_table_name}'.")
 
@@ -188,6 +136,8 @@ class Resampler:
 
                 except Exception as e:
                     logging.error(f"Failed to process coin '{coin}': {e}")
+        finally:
+            conn.close()
     
         self._log_summary()
         self._save_status()

scripts/backup_runner.py

@@ -0,0 +1,74 @@
+#!/usr/bin/env python3
+"""
+Backup Runner
+
+Creates a daily pg_dump backup of the PostgreSQL database, compresses it,
+and retains only the last 7 days of backups.
+
+Designed to run as a periodic cron job (daily) inside the Docker container.
+Backups are written to /backups which is mounted to a Synology shared folder.
+"""
+
+import argparse
+import logging
+import os
+import subprocess
+from datetime import datetime, timedelta
+
+from logging_utils import setup_logging
+
+BACKUP_DIR = "/backups"
+RETENTION_DAYS = 7
+
+
+def run_backup():
+    """Run pg_dump and compress the output."""
+    today = datetime.now().strftime("%Y%m%d")
+    backup_file = os.path.join(BACKUP_DIR, f"hyper_{today}.sql.gz")
+
+    os.makedirs(BACKUP_DIR, exist_ok=True)
+
+    logging.info(f"Starting backup to {backup_file}")
+
+    cmd = f"pg_dump -h postgres -U hyper hyper | gzip > {backup_file}"
+    result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
+
+    if result.returncode == 0:
+        file_size = os.path.getsize(backup_file)
+        logging.info(f"Backup completed: {backup_file} ({file_size:,} bytes)")
+    else:
+        logging.error(f"Backup failed: {result.stderr}")
+        if os.path.exists(backup_file):
+            os.remove(backup_file)
+
+    cleanup_old_backups()
+
+
+def cleanup_old_backups():
+    """Delete backup files older than RETENTION_DAYS."""
+    cutoff = datetime.now() - timedelta(days=RETENTION_DAYS)
+
+    if not os.path.exists(BACKUP_DIR):
+        return
+
+    for filename in os.listdir(BACKUP_DIR):
+        if filename.startswith("hyper_") and filename.endswith(".sql.gz"):
+            filepath = os.path.join(BACKUP_DIR, filename)
+            mtime = datetime.fromtimestamp(os.path.getmtime(filepath))
+            if mtime < cutoff:
+                os.remove(filepath)
+                logging.info(f"Deleted old backup: {filename}")
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Run PostgreSQL backup.")
+    parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug'])
+    args = parser.parse_args()
+
+    setup_logging(args.log_level, 'BackupRunner')
+
+    run_backup()
+
+
+if __name__ == "__main__":
+    main()

scripts/cron_scheduler.py

@@ -0,0 +1,99 @@
+#!/usr/bin/env python3
+"""
+Cron Scheduler
+
+Runs periodic maintenance tasks inside the Docker container using the
+`schedule` library. This replaces a system cron daemon and keeps all
+scheduling logic in Python.
+
+Scheduled tasks:
+  - data_fetcher.py   — daily at 02:00 UTC (full historical catch-up)
+  - fetch_history.py  — daily at 03:00 UTC (additional history fetch)
+  - gap_detector.py   — hourly at :15 (fill missing 1m candles)
+  - backup_runner.py  — daily at 04:00 UTC (pg_dump backup)
+"""
+
+import argparse
+import logging
+import os
+import subprocess
+import sys
+import time
+import schedule
+import signal
+
+from logging_utils import setup_logging
+
+shutdown_requested = False
+
+
+def handle_shutdown(signum, frame):
+    global shutdown_requested
+    shutdown_requested = True
+
+
+def run_data_fetcher():
+    try:
+        logging.info("Running data_fetcher.py")
+        subprocess.run([
+            sys.executable, "data_fetcher.py",
+            "--coins", "BTC", "ETH", "SOL", "BNB", "HYPE", "SUI",
+            "xyz:BRENTOIL", "xyz:CL", "xyz:GOLD", "xyz:SILVER",
+            "mkts:USTECH", "xyz:XYZ100",
+            "--interval", "1m", "--days", "7", "--log-level", "normal"
+        ], check=True)
+    except Exception as e:
+        logging.error(f"Data fetcher failed: {e}")
+
+
+def run_fetch_history():
+    try:
+        logging.info("Running fetch_history.py")
+        subprocess.run([sys.executable, "fetch_history.py", "--log-level", "normal"], check=True)
+    except Exception as e:
+        logging.error(f"Fetch history failed: {e}")
+
+
+def run_gap_detector():
+    try:
+        logging.info("Running gap_detector.py")
+        subprocess.run([sys.executable, "scripts/gap_detector.py", "--log-level", "normal"], check=True)
+    except Exception as e:
+        logging.error(f"Gap detector failed: {e}")
+
+
+def run_backup():
+    try:
+        logging.info("Running backup_runner.py")
+        subprocess.run([sys.executable, "scripts/backup_runner.py", "--log-level", "normal"], check=True)
+    except Exception as e:
+        logging.error(f"Backup failed: {e}")
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Run periodic maintenance tasks.")
+    parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug'])
+    args = parser.parse_args()
+
+    signal.signal(signal.SIGTERM, handle_shutdown)
+    signal.signal(signal.SIGINT, handle_shutdown)
+
+    setup_logging(args.log_level, 'CronScheduler')
+
+    # Schedule jobs
+    schedule.every().day.at("02:00").do(run_data_fetcher)
+    schedule.every().day.at("03:00").do(run_fetch_history)
+    schedule.every().hour.at(":15").do(run_gap_detector)
+    schedule.every().day.at("04:00").do(run_backup)
+
+    logging.info("Cron scheduler started")
+
+    while not shutdown_requested:
+        schedule.run_pending()
+        time.sleep(1)
+
+    logging.info("Cron scheduler shutting down.")
+
+
+if __name__ == "__main__":
+    main()

scripts/gap_detector.py

@@ -0,0 +1,137 @@
+#!/usr/bin/env python3
+"""
+Gap Detector
+
+Detects missing 1-minute candle data in the PostgreSQL database and
+backfills gaps by fetching historical data from the Hyperliquid HTTP API.
+
+Designed to run as a periodic cron job (hourly) inside the Docker container.
+"""
+
+import argparse
+import logging
+import os
+import sys
+import time
+from datetime import datetime, timedelta, timezone
+
+import pandas as pd
+from hyperliquid.info import Info
+from hyperliquid.utils import constants
+
+from logging_utils import setup_logging
+from db import get_connection, sanitize_table_name, upsert_candles
+
+WATCHED_COINS = [
+    "BTC", "ETH", "SOL", "BNB", "HYPE", "SUI",
+    "xyz:BRENTOIL", "xyz:CL", "xyz:GOLD", "xyz:SILVER",
+    "mkts:USTECH", "xyz:XYZ100"
+]
+
+
+def detect_and_fill_gaps(coin, conn):
+    """Detect gaps in the 1m data for a coin and backfill them."""
+    table_name = sanitize_table_name(coin, "1m")
+
+    now = datetime.now(timezone.utc)
+    start = now - timedelta(hours=24)
+
+    query = f'SELECT timestamp_ms FROM "{table_name}" WHERE timestamp_ms >= %s ORDER BY timestamp_ms'
+    df = pd.read_sql(query, conn, params=(int(start.timestamp() * 1000),))
+
+    if df.empty:
+        logging.info(f"No data for {coin} in the last 24 hours, skipping gap detection")
+        return
+
+    existing_timestamps = set(df['timestamp_ms'].tolist())
+
+    # Generate expected timestamps (every minute)
+    expected_timestamps = set()
+    current = start
+    while current <= now:
+        expected_timestamps.add(int(current.timestamp() * 1000))
+        current += timedelta(minutes=1)
+
+    gaps = expected_timestamps - existing_timestamps
+
+    if not gaps:
+        logging.info(f"No gaps found for {coin}")
+        return
+
+    logging.info(f"Found {len(gaps)} gaps for {coin}, backfilling...")
+
+    # Find contiguous gap ranges
+    sorted_gaps = sorted(gaps)
+    gap_ranges = []
+    gap_start = sorted_gaps[0]
+    gap_end = sorted_gaps[0]
+
+    for ts in sorted_gaps[1:]:
+        if ts == gap_end + 60000:
+            gap_end = ts
+        else:
+            gap_ranges.append((gap_start, gap_end + 60000))
+            gap_start = ts
+            gap_end = ts
+    gap_ranges.append((gap_start, gap_end + 60000))
+
+    info = Info(constants.MAINNET_API_URL, skip_ws=True)
+
+    for gap_start_ms, gap_end_ms in gap_ranges:
+        logging.info(
+            f"Backfilling gap for {coin}: "
+            f"{datetime.fromtimestamp(gap_start_ms/1000, tz=timezone.utc)} "
+            f"to {datetime.fromtimestamp(gap_end_ms/1000, tz=timezone.utc)}"
+        )
+
+        current_start = gap_start_ms
+        while current_start < gap_end_ms:
+            try:
+                batch = info.candles_snapshot(coin, "1m", current_start, gap_end_ms)
+                if not batch:
+                    break
+
+                records = []
+                for candle in batch:
+                    records.append((
+                        datetime.fromtimestamp(candle['t'] / 1000, tz=timezone.utc).strftime('%Y-%m-%d %H:%M:%S'),
+                        candle['t'],
+                        candle.get('o'), candle.get('h'), candle.get('l'), candle.get('c'),
+                        candle.get('v'), candle.get('n')
+                    ))
+
+                upsert_candles(conn, table_name, records)
+
+                last_ts = batch[-1]['t']
+                if last_ts < current_start:
+                    break
+                current_start = last_ts + 1
+                time.sleep(0.5)
+            except Exception as e:
+                logging.error(f"Error backfilling gap for {coin}: {e}")
+                break
+
+    logging.info(f"Gap backfilling complete for {coin}")
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Detect and fill gaps in 1m candle data.")
+    parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug'])
+    args = parser.parse_args()
+
+    setup_logging(args.log_level, 'GapDetector')
+
+    conn = get_connection()
+
+    for coin in WATCHED_COINS:
+        try:
+            detect_and_fill_gaps(coin, conn)
+        except Exception as e:
+            logging.error(f"Error detecting gaps for {coin}: {e}")
+
+    conn.close()
+    logging.info("Gap detection complete!")
+
+
+if __name__ == "__main__":
+    main()

scripts/resampler_loop.py

@@ -0,0 +1,69 @@
+#!/usr/bin/env python3
+"""
+Resampler Loop Wrapper
+
+Runs the Resampler in a continuous loop, executing it once per minute.
+This replaces the schedule-based approach used in main_app.py and is
+designed to run as a supervisord-managed process inside Docker.
+"""
+
+import argparse
+import logging
+import os
+import sys
+import time
+import signal
+
+from logging_utils import setup_logging
+from resampler import Resampler, parse_timeframes
+
+shutdown_requested = False
+
+
+def handle_shutdown(signum, frame):
+    global shutdown_requested
+    shutdown_requested = True
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Run the resampler in a continuous loop.")
+    parser.add_argument("--coins", nargs='+', required=True, help="List of coins to process.")
+    parser.add_argument("--timeframes", nargs='+', required=True, help="List of timeframes to generate.")
+    parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug'])
+    args = parser.parse_args()
+
+    signal.signal(signal.SIGTERM, handle_shutdown)
+    signal.signal(signal.SIGINT, handle_shutdown)
+
+    setup_logging(args.log_level, 'ResamplerLoop')
+
+    timeframes_dict = parse_timeframes(args.timeframes)
+    logging.info(f"Resampler loop started. Coins: {args.coins}, Timeframes: {list(timeframes_dict.keys())}")
+
+    while not shutdown_requested:
+        try:
+            # Pass a copy because Resampler.run() deletes '1m' from the dict
+            timeframes_copy = dict(timeframes_dict)
+            resampler = Resampler(
+                log_level=args.log_level,
+                coins=args.coins,
+                timeframes=timeframes_copy
+            )
+            resampler.run()
+        except Exception as e:
+            logging.error(f"Resampler run failed: {e}")
+
+        if shutdown_requested:
+            break
+
+        # Sleep for 60 seconds, but check shutdown flag every second
+        for _ in range(60):
+            if shutdown_requested:
+                break
+            time.sleep(1)
+
+    logging.info("Resampler loop shutting down.")
+
+
+if __name__ == "__main__":
+    main()

strategies/base_strategy.py

@@ -4,7 +4,7 @@ import json
 import os
 import logging
 from datetime import datetime, timezone
-import sqlite3
+import psycopg2
 import multiprocessing
 import time
 
@@ -27,7 +27,7 @@ class BaseStrategy(ABC):
         
         self.coin = params.get("coin", "N/A")
         self.timeframe = params.get("timeframe", "N/A")
-        self.db_path = os.path.join("_data", "market_data.db")
+        self.db_path = os.environ.get("PG_CONN_STR", "postgresql://hyper:hyper@localhost:5432/hyper")
         self.status_file_path = os.path.join("_data", f"strategy_status_{self.strategy_name}.json")
 
         self.current_signal = "INIT"
@@ -38,19 +38,23 @@ class BaseStrategy(ABC):
 
     def load_data(self) -> pd.DataFrame:
         """Loads historical data for the configured coin and timeframe."""
-        table_name = f"{self.coin}_{self.timeframe}"
+        table_name = f"{self.coin.replace(':', '_')}_{self.timeframe}"
         
         periods = [v for k, v in self.params.items() if 'period' in k or '_ma' in k or 'slow' in k or 'fast' in k]
         limit = max(periods) + 50 if periods else 500
 
         try:
-            with sqlite3.connect(f"file:{self.db_path}?mode=ro", uri=True) as conn:
+            conn = psycopg2.connect(self.db_path)
+            conn.set_session(readonly=True)
+            try:
                 query = f'SELECT * FROM "{table_name}" ORDER BY datetime_utc DESC LIMIT {limit}'
                 df = pd.read_sql(query, conn, parse_dates=['datetime_utc'])
                 if df.empty: return pd.DataFrame()
                 df.set_index('datetime_utc', inplace=True)
                 df.sort_index(inplace=True)
                 return df
+            finally:
+                conn.close()
         except Exception as e:
             logging.error(f"Failed to load data from table '{table_name}': {e}")
             return pd.DataFrame()

supervisord.conf

@@ -0,0 +1,38 @@
+[supervisord]
+nodaemon=true
+
+[program:live_candle_fetcher]
+command=python live_candle_fetcher.py --coins BTC ETH SOL BNB HYPE SUI xyz:BRENTOIL xyz:CL xyz:GOLD xyz:SILVER mkts:USTECH xyz:XYZ100 --log-level normal
+autostart=true
+autorestart=true
+stdout_logfile=/app/_logs/live_candle_fetcher.log
+stderr_logfile=/app/_logs/live_candle_fetcher.log
+stdout_logfile_maxbytes=10MB
+stdout_logfile_backups=3
+
+[program:resampler_loop]
+command=python scripts/resampler_loop.py --coins BTC ETH SOL BNB HYPE SUI xyz:BRENTOIL xyz:CL xyz:GOLD xyz:SILVER mkts:USTECH xyz:XYZ100 --timeframes 3m 5m 15m 30m 1h 2h 4h 8h 12h 1d 3d 1w 1M 148m 37m --log-level normal
+autostart=true
+autorestart=true
+stdout_logfile=/app/_logs/resampler.log
+stderr_logfile=/app/_logs/resampler.log
+stdout_logfile_maxbytes=10MB
+stdout_logfile_backups=3
+
+[program:indicators_fetcher]
+command=python indicators_fetcher.py --log-level normal
+autostart=true
+autorestart=true
+stdout_logfile=/app/_logs/indicators_fetcher.log
+stderr_logfile=/app/_logs/indicators_fetcher.log
+stdout_logfile_maxbytes=10MB
+stdout_logfile_backups=3
+
+[program:cron_scheduler]
+command=python scripts/cron_scheduler.py --log-level normal
+autostart=true
+autorestart=true
+stdout_logfile=/app/_logs/cron_scheduler.log
+stderr_logfile=/app/_logs/cron_scheduler.log
+stdout_logfile_maxbytes=10MB
+stdout_logfile_backups=3