p1otek/hyper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add indicators fetcher, rich dashboard renderer, and remove trade executor/status

Browse Source
This commit is contained in:
DiTus
2026-07-29 09:11:13 +02:00
parent 2a8ee9c8c5
commit 63bab43557
7 changed files with 1166 additions and 187 deletions
Download Patch File Download Diff File Expand all files Collapse all files

90
WIKI/dashboard_configuration.md Normal file
View File

@ -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 |

239
WIKI/indicators.md Normal file
View File

@ -0,0 +1,239 @@
# 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
}
```
- **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
### `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
## 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.

254
dashboard.py Normal file
View File

@ -0,0 +1,254 @@
"""
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, Group
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,
}
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 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_layout(self, watched_coins, prices, display_names, strategy_statuses, strategy_configs, indicators_status=None):
"""Build the complete dashboard layout with vertically stacked tables."""
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), (2, 0, 0, 0)))
if self.table_visibility.get("strategies", True):
tables.append(self.build_strategy_table(strategy_statuses, strategy_configs))
if not tables:
return Layout()
return Layout(Group(*tables))

425
indicators.py Normal file
View File

@ -0,0 +1,425 @@
"""
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 sqlite3
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}_{timeframe}"
try:
with sqlite3.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}_{timeframe}"
try:
with sqlite3.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}_{timeframe}"
try:
with sqlite3.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}_{timeframe}"
den_table = f"{den_coin}_{timeframe}"
try:
with sqlite3.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}_{timeframe}"
den_table = f"{den_coin}_{timeframe}"
try:
with sqlite3.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}_{timeframe}"
den_table = f"{den_coin}_{timeframe}"
try:
with sqlite3.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:
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:
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:
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:
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

86
indicators_fetcher.py Normal file
View File

@ -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.path.join(project_root, "_data", "market_data.db")
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.")

258
main_app.py
View File

@ -8,7 +8,7 @@ import multiprocessing
import schedule
import sqlite3
import pandas as pd
from datetime import datetime, timezone
from datetime import datetime
import importlib
# --- REMOVED: import signal ---
# --- REMOVED: from queue import Empty ---
@ -18,6 +18,9 @@ from logging_utils import setup_logging
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
# --- Rich dashboard renderer ---
from dashboard import DashboardRenderer
from rich.live import Live
# --- Configuration ---
WATCHED_COINS = ["BTC", "ETH", "SOL", "BNB", "HYPE", "SUI", "xyz:BRENTOIL", "xyz:CL"]
@ -31,24 +34,11 @@ 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 +338,53 @@ 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.renderer = DashboardRenderer(table_visibility={
"market": True,
"strategies": False,
"indicators": True,
})
def read_prices(self):
"""Reads the latest prices directly from the shared memory dictionary."""
@ -386,190 +412,47 @@ class MainApp:
enabled_statuses[name] = {"current_signal": "Initializing..."}
self.strategy_statuses = enabled_statuses
def read_executor_status(self):
"""Reads the live status file from the trade executor."""
if os.path.exists(TRADE_EXECUTOR_STATUS_FILE):
def read_indicators_status(self):
"""Reads the indicators status JSON 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:
# --- FIX: Read the 'open_positions' key from the file ---
status_data = json.load(f)
self.open_positions = status_data.get('open_positions', {})
with open(status_file, 'r', encoding='utf-8') as f:
self.indicators_status = json.load(f)
except (IOError, json.JSONDecodeError):
logging.debug("Could not read trade executor status file.")
self.indicators_status = {}
else:
self.open_positions = {}
self.indicators_status = {}
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):
"""Helper function to format prices for the dashboard."""
try:
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 toggle_table(self, table_name, enabled=None):
"""Toggle a dashboard table's visibility at runtime."""
return self.renderer.toggle_table(table_name, enabled)
def display_dashboard(self):
"""Displays a formatted dashboard with side-by-side tables."""
print("\x1b[H\x1b[J", end="") # Clear screen
left_table_lines = ["--- Market Dashboard ---"]
# --- MODIFIED: Adjusted width for new columns ---
left_table_width = 65
left_table_lines.append("-" * left_table_width)
# --- MODIFIED: Replaced Market Cap with Gap ---
left_table_lines.append(f"{'#':<2} | {'Coin':^6} | {'Best Bid':>10} | {'Live Price':>10} | {'Best Ask':>10} | {'Gap':>10} |")
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()
"""Build and return the rich dashboard layout."""
return self.renderer.build_layout(
self.watched_coins,
self.prices,
COIN_DISPLAY_NAMES,
self.strategy_statuses,
self.strategy_configs,
self.indicators_status
)
def run(self):
"""Main loop to read data, display dashboard, and check processes."""
while True:
self.read_prices()
# --- REMOVED: self.read_market_caps() ---
self.read_strategy_statuses()
self.read_executor_status()
# --- REMOVED: self.check_process_status() ---
self.display_dashboard()
time.sleep(0.5)
with Live(self.display_dashboard(), refresh_per_second=2, console=self.renderer.console) as live:
while True:
self.read_prices()
self.read_strategy_statuses()
self.read_indicators_status()
live.update(self.display_dashboard())
time.sleep(0.5)
if __name__ == "__main__":
setup_logging('normal', 'MainApp')
@ -613,6 +496,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,

1
requirements.txt
View File

@ -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