Clean up unused files, organize structure, update docs

- Delete obsolete files: data_fetcher_old.py, market_old.py, base_strategy.py (root),
  strategy_sma_cross.py, and old architecture remnants (address_monitor.py,
  position_monitor.py, trade_log.py, wallet_data.py, whale_tracker.py)
- Delete zero-byte Docker artifacts and runtime files (clp_hedger.log,
  clp_hedger/hedge_status.json)
- Move one-off utility scripts to scripts/ directory
- Move example/template files to .temp/ directory
- Update .gitignore: add entries for clp_hedger.log, clp_hedger/hedge_status.json,
  Docker layer hash files, Using, Running, and backups/
- Update .dockerignore: add clp_hedger.log, clp_hedger/hedge_status.json, backups/
- Create example config files: _data/strategies.json.example,
  _data/backtesting_conf.json.example, _data/coin_precision.json.example
- Update GEMINI.md: remove outdated session summaries and duplicate review section
- Update review.md: add cleanup status section, update remaining recommendations
- Update MIGRATION_PLAN.md: mark completed phases, update file references
- Update DOCKER_MIGRATION_GUIDE.md: update import_csv.py path reference
This commit is contained in:
DiTus
2026-08-05 09:50:36 +02:00
parent 967c86e8e9
commit 1a95fe1caa
39 changed files with 599 additions and 3116 deletions

View File

@ -12,3 +12,6 @@ agents/
secrets/ secrets/
.env.docker .env.docker
.env .env
clp_hedger.log
clp_hedger/hedge_status.json
backups/

12
.gitignore vendored
View File

@ -22,6 +22,9 @@ _data/*.json
# Ignore all log files # Ignore all log files
_logs/ _logs/
# Ignore backups
backups/
# --- SDK --- # --- SDK ---
# Ignore all contents of the sdk directory # Ignore all contents of the sdk directory
sdk/ sdk/
@ -30,6 +33,15 @@ sdk/
# Ignore custom agents directory # Ignore custom agents directory
agents/ agents/
# Ignore CLP hedger runtime log and status
clp_hedger.log
clp_hedger/hedge_status.json
# Ignore Docker layer hash files and artifacts
/[0-9a-f]{12}
/Running
/Using
# Ignore temporary files and examples # Ignore temporary files and examples
.temp/ .temp/

247
DOCKER_MIGRATION_GUIDE.md Normal file
View File

@ -0,0 +1,247 @@
# Docker Image Build & SQLite-to-PostgreSQL Migration Guide
## Prerequisites
- Docker and Docker Compose installed on the host
- Access to the `hyper` project directory
- SQLite database file (`_data/market_data.db`)
- `.env.docker` file with PostgreSQL credentials (see `.env.docker.example`)
## 1. Building / Rebuilding Docker Images
The `data-collector` image is built from `Dockerfile` and contains all Python source
files. Whenever source code changes (e.g., `db.py`, `migrate_sqlite_to_pg.py`),
**you must rebuild the image** — the container does not mount source files from
the host.
### Build Command
```bash
docker build --network host -t hyper-data-collector:latest .
```
> **Note:** `--network host` is used to speed up `pip install` by avoiding Docker's
> default bridge network. Omit it if building on a system where host networking is
> not available.
### Rebuild Checklist
1. Make code changes in the project directory
2. Rebuild the image: `docker build -t hyper-data-collector:latest .`
3. Restart containers: `docker-compose down && docker-compose up -d`
4. Wait for PostgreSQL healthcheck:
```bash
until docker-compose exec postgres pg_isready -U hyper -d hyper; do sleep 2; done
```
## 2. Running the Migration
### Step 1: Ensure Containers Are Running
```bash
docker-compose up -d
```
Wait for PostgreSQL to be ready:
```bash
until docker-compose exec postgres pg_isready -U hyper -d hyper; do sleep 2; done
```
### Step 2: Run the Migration Script
```bash
docker-compose run --rm data-collector \
python migrate_sqlite_to_pg.py \
--sqlite-path _data/market_data.db \
--log-level normal
```
**Arguments:**
| Argument | Description | Default |
|---|---|---|
| `--sqlite-path` | Path to the SQLite database file | `_data/market_data.db` |
| `--log-level` | Logging verbosity: `off`, `normal`, `debug` | `normal` |
### Step 3: Verify Migration
Check row counts in PostgreSQL:
```bash
docker-compose exec postgres psql -U hyper -d hyper -c \
"SELECT COUNT(*) FROM \"0G_1m\";"
```
Compare with the source SQLite row count:
```bash
sqlite3 _data/market_data.db "SELECT COUNT(*) FROM \"0G_1m\";"
```
## 3. Migration Process Details
The migration script (`migrate_sqlite_to_pg.py`) performs the following:
1. **Connects** to SQLite (source) and PostgreSQL (destination)
2. **Enumerates** all candle tables by matching suffixes (`_1m`, `_3m`, `_5m`, etc.)
3. **Skips** legacy tables (`market_cap`, `candles`, `daily`)
4. **For each table:**
- Reads all rows from SQLite via `pandas.read_sql`
- Creates the PostgreSQL table if it doesn't exist (`db.create_candle_table`)
- **Deduplicates** records by `timestamp_ms` (handles duplicate timestamps in source)
- Batch-upserts records using `execute_values` with `ON CONFLICT DO UPDATE`
5. **Commits** after each table
### Tables Migrated
Tables are identified by their timeframe suffix. Supported timeframes:
```
1m, 3m, 5m, 15m, 30m, 37m, 148m, 1h, 2h, 4h, 8h, 12h, 1d, 3d, 1w, 1month
```
### Table Name Sanitization
Coin symbols containing colons (e.g., `xyz:BRENTOIL`) are sanitized to
`xyz_BRENTOIL` for PostgreSQL compatibility.
### Deduplication
SQLite databases may contain duplicate `timestamp_ms` entries within the same
table. The `upsert_candles` function in `db.py` deduplicates records by
`timestamp_ms` before batch insertion, keeping the last occurrence. This prevents
PostgreSQL's `ON CONFLICT` cardinality violation:
```
ON CONFLICT DO UPDATE command cannot affect row a second time
```
## 4. Re-running the Migration
The migration is **idempotent** — you can safely re-run it:
- `CREATE TABLE IF NOT EXISTS` skips existing tables
- `ON CONFLICT DO UPDATE` overwrites existing rows with the same `timestamp_ms`
- Deduplication ensures no cardinality errors on re-runs
To re-run after code changes:
```bash
docker build -t hyper-data-collector:latest .
docker-compose down
docker-compose up -d
until docker-compose exec postgres pg_isready -U hyper -d hyper; do sleep 2; done
docker-compose run --rm data-collector \
python migrate_sqlite_to_pg.py \
--sqlite-path _data/market_data.db \
--log-level normal
```
## 5. Troubleshooting
### `ON CONFLICT DO UPDATE command cannot affect row a second time`
**Cause:** Duplicate `timestamp_ms` values in the same batch being inserted.
**Fix:** The `upsert_candles` function in `db.py` deduplicates records by
`timestamp_ms` before insertion. Ensure you're running the latest image:
```bash
docker build -t hyper-data-collector:latest .
```
### `connection to server at "postgres" failed: Connection timed out`
**Cause:** PostgreSQL container is not ready or not running.
**Fix:** Wait for the healthcheck to pass before running the migration:
```bash
until docker-compose exec postgres pg_isready -U hyper -d hyper; do sleep 2; done
```
### `SQLite database not found at '_data/market_data.db'`
**Cause:** The SQLite database file is not mounted into the container.
**Fix:** Ensure the `_data` volume is mounted in `docker-compose.yml`:
```yaml
volumes:
- ./_data:/app/_data
```
And the database file exists on the host:
```bash
ls -la _data/market_data.db
```
### Migration is slow (12+ hours for large databases)
**Tips:**
- Use `--log-level off` to reduce I/O from logging
- The `page_size=1000` in `execute_values` is already optimal
- Ensure PostgreSQL has adequate `shared_buffers` (see `postgres/postgresql.conf`)
### `psycopg2.errors.DuplicateTable` or table already exists
**Cause:** The table was partially migrated in a previous run.
**Fix:** This is handled gracefully by `CREATE TABLE IF NOT EXISTS`. The migration
will continue from where it left off. Re-run the migration script.
### Container exits immediately after `docker-compose up -d`
**Cause:** Missing `.env.docker` file or missing secrets.
**Fix:**
```bash
cp .env.docker.example .env.docker
cp secrets/pg_password.txt.example secrets/pg_password.txt
```
Edit `.env.docker` to set the correct `PG_CONN_STR` if needed.
### Check Container Logs
```bash
# PostgreSQL logs
docker-compose logs postgres
# Data collector logs
docker-compose logs data-collector
# Follow logs in real-time
docker-compose logs -f
```
### Verify PostgreSQL Data
```bash
# List all tables
docker-compose exec postgres psql -U hyper -d hyper -c \
"\dt"
# Check row count for a specific table
docker-compose exec postgres psql -U hyper -d hyper -c \
"SELECT COUNT(*) FROM \"BTC_1m\";"
# Check for data gaps
docker-compose exec postgres psql -U hyper -d hyper -c \
"SELECT datetime_utc FROM \"BTC_1m\" ORDER BY timestamp_ms LIMIT 5;"
```
## 6. Post-Migration
After migration completes successfully:
1. **Update host applications** to connect to `localhost:5432` instead of SQLite
2. **Start the data collector** for ongoing data collection:
```bash
docker-compose up -d
```
3. **Set up backups** using the backup runner script
4. **Monitor** the gap detector for any missing data

198
GEMINI.md
View File

@ -46,201 +46,3 @@ python main_app.py
* **Strategies:** Custom strategies should inherit from the `BaseStrategy` class (defined in `strategies/base_strategy.py`) and implement the `calculate_signals` method. * **Strategies:** Custom strategies should inherit from the `BaseStrategy` class (defined in `strategies/base_strategy.py`) and implement the `calculate_signals` method.
* **Documentation:** The `WIKI/` directory contains detailed documentation for the project. Start with `WIKI/SUMMARY.md`. * **Documentation:** The `WIKI/` directory contains detailed documentation for the project. Start with `WIKI/SUMMARY.md`.
## Session Summary
**Date:** 2025-11-10
**Objective(s):**
Fix urllib3 SSL compatibility warning and create sessionsummary agent following OpenCode.ai guidelines
**Key Accomplishments:**
* Resolved NotOpenSSLWarning by downgrading urllib3 from 2.5.0 to 1.26.20
* Updated requirements.txt to prevent future SSL compatibility issues
* Created sessionsummary agent in .opencode/agent/ following OpenCode.ai specifications
* Removed incorrect Python implementation and created proper markdown agent configuration
**Decisions Made:**
* Chose to downgrade urllib3 instead of upgrading SSL environment for stability
* Followed OpenCode.ai agent guidelines instead of creating custom Python implementation
* Configured sessionsummary as subagent with proper permissions and tools
**Key Files Modified:**
* `requirements.txt`
* `GEMINI.md`
* `.opencode/agent/sessionsummary.md`
**Next Steps/Open Questions:**
* Test trading bot functionality after SSL fix to ensure no regressions
* Integrate sessionsummary agent into regular development workflow
* Add .opencode/ to .gitignore if not already present
## Session Summary
**Date:** 2025-11-11
**Objective(s):**
Start new Gemini session and organize project files by creating .temp folder for examples and temporary files
**Key Accomplishments:**
* Created .temp folder for organizing examples and temporary files
* Updated .gitignore to include .temp/ directory
* Moved model_comparison_examples.md to .temp folder for better organization
* Established file management practices for future development
**Decisions Made:**
* Chose to use .temp folder instead of mixing examples with main project files
* Added .temp to .gitignore to prevent accidental commits of temporary files
* Followed user instruction to organize project structure for better maintainability
**Key Files Modified:**
* `.gitignore`
* `.temp/` (created)
* `model_comparison_examples.md` (moved to .temp/)
**Next Steps/Open Questions:**
* Continue organizing any other example or temporary files into .temp folder
* Maintain consistent file organization practices in future development
* Consider creating additional organizational directories if needed
## Session Summary
**Date:** 2025-11-11
**Objective(s):**
Fix DashboardDataFetcher path resolution error causing file operation failures
**Key Accomplishments:**
* Identified root cause of file path error in dashboard_data_fetcher.py subprocess execution
* Fixed path resolution by using absolute paths instead of relative paths
* Added os.makedirs() call to ensure _logs directory exists before file operations
* Tested fix and confirmed DashboardDataFetcher now works correctly
* Committed and pushed fix to remote repository
**Decisions Made:**
* Used os.path.dirname(os.path.abspath(__file__)) to get correct project root
* Ensured backward compatibility while fixing the path resolution issue
* Maintained atomic file write pattern for data integrity
**Key Files Modified:**
* `dashboard_data_fetcher.py`
* `GEMINI.md`
**Next Steps/Open Questions:**
* Monitor DashboardDataFetcher to ensure no further path-related errors occur
* Consider reviewing other subprocess scripts for similar path resolution issues
* Test main_app.py to ensure dashboard displays data correctly
## Session Summary
**Date:** 2025-11-11
**Objective(s):**
Debug and fix DashboardDataFetcher path resolution error causing file operation failures
**Key Accomplishments:**
* Identified root cause of file path error in dashboard_data_fetcher.py subprocess execution
* Fixed path resolution by using absolute paths instead of relative paths
* Added os.makedirs() call to ensure _logs directory exists before file operations
* Tested fix and confirmed DashboardDataFetcher now works correctly
* Committed and pushed fix to remote repository
* Organized project files with .temp folder for better structure
**Decisions Made:**
* Used os.path.dirname(os.path.abspath(__file__)) to get correct project root
* Ensured backward compatibility while fixing path resolution issue
* Maintained atomic file write pattern for data integrity
* Added proper directory existence checks to prevent runtime errors
**Key Files Modified:**
* `dashboard_data_fetcher.py`
* `GEMINI.md`
* `.gitignore`
* `.temp/` (created)
**Next Steps/Open Questions:**
* Monitor DashboardDataFetcher to ensure no further path-related errors occur
* Consider reviewing other subprocess scripts for similar path resolution issues
* Test main_app.py to ensure dashboard displays data correctly
* Continue improving project organization and file management practices
---
# Project Review and Recommendations
This review provides an analysis of the current state of the automated trading bot project, proposes specific code improvements, and identifies files that appear to be unused or are one-off utilities that could be reorganized.
The project is a well-structured, multi-process Python application for crypto trading. It has a clear separation of concerns between data fetching, strategy execution, and trade management. The use of `multiprocessing` and a centralized `main_app.py` orchestrator is a solid architectural choice.
The following sections detail recommendations for improving configuration management, code structure, and robustness, along with a list of files recommended for cleanup.
---
## Proposed Code Changes
### 1. Centralize Configuration
- **Issue:** Key configuration variables like `WATCHED_COINS` and `required_timeframes` are hardcoded in `main_app.py`. This makes them difficult to change without modifying the source code.
- **Proposal:**
- Create a central configuration file, e.g., `_data/config.json`.
- Move `WATCHED_COINS` and `required_timeframes` into this new file.
- Load this configuration in `main_app.py` at startup.
- **Benefit:** Decouples configuration from code, making the application more flexible and easier to manage.
### 2. Refactor `main_app.py` for Clarity
- **Issue:** `main_app.py` is long and handles multiple responsibilities: process orchestration, dashboard rendering, and data reading.
- **Proposal:**
- **Abstract Process Management:** The functions for running subprocesses (e.g., `run_live_candle_fetcher`, `run_resampler_job`) contain repetitive logic for logging, shutdown handling, and process looping. This could be abstracted into a generic `ProcessRunner` class.
- **Create a Dashboard Class:** The complex dashboard rendering logic could be moved into a separate `Dashboard` class to improve separation of concerns and make the main application loop cleaner.
- **Benefit:** Improves code readability, reduces duplication, and makes the application easier to maintain and extend.
### 3. Improve Project Structure
- **Issue:** The root directory is cluttered with numerous Python scripts, making it difficult to distinguish between core application files, utility scripts, and old/example files.
- **Proposal:**
- Create a `scripts/` directory and move all one-off utility and maintenance scripts into it.
- Consider creating a `src/` or `app/` directory to house the core application source code (`main_app.py`, `trade_executor.py`, etc.), separating it clearly from configuration, data, and documentation.
- **Benefit:** A cleaner, more organized project structure that is easier for new developers to understand.
### 4. Enhance Robustness and Error Handling
- **Issue:** The agent loading in `trade_executor.py` relies on discovering environment variables by a naming convention (`_AGENT_PK`). This is clever but can be brittle if environment variables are named incorrectly.
- **Proposal:**
- Explicitly define the agent names and their corresponding environment variable keys in the proposed `_data/config.json` file. The `trade_executor` would then load only the agents specified in the configuration.
- **Benefit:** Makes agent configuration more explicit and less prone to errors from stray environment variables.
---
## Identified Unused/Utility Files
The following files were identified as likely being unused by the core application, being obsolete, or serving as one-off utilities. It is recommended to **move them to a `scripts/` directory** or **delete them** if they are obsolete.
### Obsolete / Old Versions:
- `data_fetcher_old.py`
- `market_old.py`
- `base_strategy.py` (The one in the root directory; the one in `strategies/` is used).
### One-Off Utility Scripts (Recommend moving to `scripts/`):
- `!migrate_to_sqlite.py`
- `import_csv.py`
- `del_market_cap_tables.py`
- `fix_timestamps.py`
- `list_coins.py`
- `create_agent.py`
### Examples / Unused Code:
- `basic_ws.py` (Appears to be an example file).
- `backtester.py`
- `strategy_sma_cross.py` (A strategy file in the root, not in the `strategies` folder).
- `strategy_template.py`
### Standalone / Potentially Unused Core Files:
The following files seem to have their logic already integrated into the main multi-process application. They might be remnants of a previous architecture and may not be needed as standalone scripts.
- `address_monitor.py`
- `position_monitor.py`
- `trade_log.py`
- `wallet_data.py`
- `whale_tracker.py`
### Data / Log Files (Recommend archiving or deleting):
- `hyperliquid_wallet_data_*.json` (These appear to be backups or logs).

View File

@ -34,7 +34,7 @@
│ └──────────────────────────────┘ │ │ └──────────────────────────────┘ │
└─────────────────────────────────────────────────────┘ └─────────────────────────────────────────────────────┘
Host Machine: indicators.py, base_strategy.py, main_app.py Host Machine: indicators.py, strategies/base_strategy.py, main_app.py
→ connect to localhost:5432 → connect to localhost:5432
``` ```
@ -79,9 +79,9 @@ wal_buffers = 4MB
2. `resampler.py``sqlite3``db.py` 2. `resampler.py``sqlite3``db.py`
3. `data_fetcher.py``sqlite3``db.py` 3. `data_fetcher.py``sqlite3``db.py`
4. `fetch_history.py``sqlite3``db.py` 4. `fetch_history.py``sqlite3``db.py`
5. `import_csv.py``sqlite3``db.py` 5. `scripts/import_csv.py``sqlite3``db.py`
6. `indicators.py``sqlite3``psycopg2` 6. `indicators.py``sqlite3``psycopg2`
7. `base_strategy.py``sqlite3``psycopg2` 7. `strategies/base_strategy.py``sqlite3``psycopg2`
## TODO List ## TODO List
@ -90,34 +90,34 @@ wal_buffers = 4MB
- [x] Add `psycopg2-binary` to `requirements.txt` - [x] Add `psycopg2-binary` to `requirements.txt`
### Phase 2: Modify Data Collection Components ### 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 - [x] 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` - [x] 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()` - [x] Modify `data_fetcher.py` — replace `sqlite3` with `db.py`, `to_sql()``db.upsert_candles()`
- [ ] Modify `fetch_history.py` — replace `sqlite3` with `db.py` - [x] Modify `fetch_history.py` — replace `sqlite3` with `db.py`
- [ ] Modify `import_csv.py` — replace `sqlite3` with `db.py`, `to_sql()``db.upsert_candles()` - [x] Modify `import_csv.py` — replace `sqlite3` with `db.py`, `to_sql()``db.upsert_candles()` (moved to `scripts/import_csv.py`)
### Phase 3: New Components ### Phase 3: New Components
- [ ] Create `scripts/resampler_loop.py` — wraps resampler in a while loop with 60s sleep - [x] 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 - [x] 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 - [x] 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 - [x] Create `scripts/cron_scheduler.py` — schedules data_fetcher, fetch_history, gap_detector, backup
### Phase 4: Docker Setup ### Phase 4: Docker Setup
- [ ] Create `Dockerfile` (python:3.11-slim + supervisor + psycopg2-binary) - [x] Create `Dockerfile` (python:3.11-slim + supervisor + psycopg2-binary)
- [ ] Create `docker-compose.yml` (postgres + data-collector services) - [x] Create `docker-compose.yml` (postgres + data-collector services)
- [ ] Create `supervisord.conf` (live_candle_fetcher, resampler_loop, indicators_fetcher, cron_scheduler) - [x] Create `supervisord.conf` (live_candle_fetcher, resampler_loop, indicators_fetcher, cron_scheduler)
- [ ] Create `postgres/postgresql.conf` (tuned for 4GB RAM) - [x] Create `postgres/postgresql.conf` (tuned for 4GB RAM)
- [ ] Create `.dockerignore` - [x] Create `.dockerignore`
- [ ] Create `.env.docker.example` - [x] Create `.env.docker.example`
- [ ] Create `secrets/pg_password.txt.example` - [x] Create `secrets/pg_password.txt.example`
- [ ] Update `.gitignore` - [x] Update `.gitignore`
### Phase 5: Host-Side Updates ### Phase 5: Host-Side Updates
- [ ] Modify `indicators.py` on host — connect to `localhost:5432` - [ ] Modify `indicators.py` on host — connect to `localhost:5432`
- [ ] Modify `base_strategy.py` on host — connect to `localhost:5432` - [ ] Modify `strategies/base_strategy.py` on host — connect to `localhost:5432`
### Phase 6: Migration Tool ### Phase 6: Migration Tool
- [ ] Create `migrate_sqlite_to_pg.py` — reads from SQLite, writes to PostgreSQL - [x] Create `migrate_sqlite_to_pg.py` — reads from SQLite, writes to PostgreSQL
### Phase 7: Testing & Deployment ### Phase 7: Testing & Deployment
- [ ] Commit and push to remote - [ ] Commit and push to remote

View File

@ -0,0 +1,18 @@
{
"sma_cross_eth_5m": {
"strategy_name": "sma_cross_1",
"script": "strategies.ma_cross_strategy.MaCrossStrategy",
"optimization_params": {
"fast": {
"start": 5,
"end": 150,
"step": 1
},
"slow": {
"start": 0,
"end": 0,
"step": 1
}
}
}
}

View File

@ -4,6 +4,7 @@
"AAVE": 2, "AAVE": 2,
"ACE": 2, "ACE": 2,
"ADA": 0, "ADA": 0,
"AERO": 0,
"AI": 1, "AI": 1,
"AI16Z": 1, "AI16Z": 1,
"AIXBT": 0, "AIXBT": 0,
@ -20,11 +21,11 @@
"ATOM": 2, "ATOM": 2,
"AVAX": 2, "AVAX": 2,
"AVNT": 0, "AVNT": 0,
"AXS": 1,
"AZTEC": 0,
"BABY": 0, "BABY": 0,
"BADGER": 1, "BADGER": 1,
"BANANA": 1, "BANANA": 1,
"BASH": 0,
"BATH": 0,
"BCH": 3, "BCH": 3,
"BERA": 1, "BERA": 1,
"BIGTIME": 0, "BIGTIME": 0,
@ -40,13 +41,17 @@
"BTC": 5, "BTC": 5,
"CAKE": 1, "CAKE": 1,
"CANTO": 0, "CANTO": 0,
"CASHCAT": 0,
"CATI": 0, "CATI": 0,
"CC": 0,
"CELO": 0, "CELO": 0,
"CFX": 0, "CFX": 0,
"CHILLGUY": 0, "CHILLGUY": 0,
"CHIP": 0,
"COMP": 2, "COMP": 2,
"CRV": 1, "CRV": 1,
"CYBER": 1, "CYBER": 1,
"DASH": 2,
"DOGE": 0, "DOGE": 0,
"DOOD": 0, "DOOD": 0,
"DOT": 1, "DOT": 1,
@ -61,6 +66,7 @@
"FARTCOIN": 1, "FARTCOIN": 1,
"FET": 0, "FET": 0,
"FIL": 1, "FIL": 1,
"FOGO": 0,
"FRIEND": 1, "FRIEND": 1,
"FTM": 0, "FTM": 0,
"FTT": 1, "FTT": 1,
@ -70,6 +76,7 @@
"GMT": 0, "GMT": 0,
"GMX": 2, "GMX": 2,
"GOAT": 0, "GOAT": 0,
"GRAM": 0,
"GRASS": 1, "GRASS": 1,
"GRIFFAIN": 0, "GRIFFAIN": 0,
"HBAR": 0, "HBAR": 0,
@ -78,6 +85,7 @@
"HPOS": 0, "HPOS": 0,
"HYPE": 2, "HYPE": 2,
"HYPER": 0, "HYPER": 0,
"ICP": 1,
"ILV": 2, "ILV": 2,
"IMX": 1, "IMX": 1,
"INIT": 0, "INIT": 0,
@ -96,6 +104,7 @@
"LINEA": 0, "LINEA": 0,
"LINK": 1, "LINK": 1,
"LISTA": 0, "LISTA": 0,
"LIT": 0,
"LOOM": 0, "LOOM": 0,
"LTC": 2, "LTC": 2,
"MANTA": 1, "MANTA": 1,
@ -163,11 +172,13 @@
"SCR": 1, "SCR": 1,
"SEI": 0, "SEI": 0,
"SHIA": 0, "SHIA": 0,
"SKR": 0,
"SKY": 0, "SKY": 0,
"SNX": 1, "SNX": 1,
"SOL": 2, "SOL": 2,
"SOPH": 0, "SOPH": 0,
"SPX": 1, "SPX": 1,
"STABLE": 0,
"STBL": 0, "STBL": 0,
"STG": 0, "STG": 0,
"STRAX": 0, "STRAX": 0,
@ -201,10 +212,9 @@
"WLFI": 0, "WLFI": 0,
"XAI": 1, "XAI": 1,
"XLM": 0, "XLM": 0,
"XMR": 3,
"XPL": 0, "XPL": 0,
"XRP": 0, "XRP": 0,
"XYZ:CLUSD": 2,
"xyz:BRENTOIL": 2,
"YGG": 0, "YGG": 0,
"YZY": 0, "YZY": 0,
"ZEC": 2, "ZEC": 2,
@ -220,6 +230,5 @@
"kLUNC": 0, "kLUNC": 0,
"kNEIRO": 1, "kNEIRO": 1,
"kPEPE": 0, "kPEPE": 0,
"kSHIB": 0, "kSHIB": 0
"xyz:CLUSD": 2
} }

View File

@ -0,0 +1,10 @@
{
"BTC": 5,
"ETH": 4,
"SOL": 2,
"BNB": 3,
"HYPE": 2,
"SUI": 1,
"0G": 0,
"2Z": 0
}

View File

@ -0,0 +1,50 @@
{
"sma_cross_1": {
"enabled": false,
"class": "strategies.ma_cross_strategy.MaCrossStrategy",
"agent": "scalper_agent",
"parameters": {
"coin": "ETH",
"timeframe": "15m",
"short_ma": 7,
"long_ma": 44,
"size": 0.0055,
"leverage_long": 5,
"leverage_short": 5
}
},
"sma_44d_btc": {
"enabled": false,
"class": "strategies.single_sma_strategy.SingleSmaStrategy",
"parameters": {
"agent": "swing",
"coin": "BTC",
"timeframe": "1d",
"sma_period": 44,
"size": 0.0001,
"leverage_long": 3,
"leverage_short": 1
}
},
"copy_trader_eth": {
"enabled": true,
"is_event_driven": true,
"class": "strategies.copy_trader_strategy.CopyTraderStrategy",
"parameters": {
"agent": "scalper",
"target_address": "0x32885a6adac4375858E6edC092EfDDb0Ef46484C",
"coins_to_copy": {
"ETH": {
"size": 0.0055,
"leverage_long": 3,
"leverage_short": 3
},
"BTC": {
"size": 0.0002,
"leverage_long": 1,
"leverage_short": 1
}
}
}
}
}

View File

@ -1,221 +0,0 @@
import os
import sys
import time
import json
import argparse
from datetime import datetime, timezone
from hyperliquid.info import Info
from hyperliquid.utils import constants
from collections import deque
import logging
import csv
from logging_utils import setup_logging
# --- Configuration ---
DEFAULT_ADDRESSES_TO_WATCH = [
#"0xd4c1f7e8d876c4749228d515473d36f919583d1d",
"0x47930c76790c865217472f2ddb4d14c640ee450a",
# "0x4d69495d16fab95c3c27b76978affa50301079d0",
# "0x09bc1cf4d9f0b59e1425a8fde4d4b1f7d3c9410d",
"0xc6ac58a7a63339898aeda32499a8238a46d88e84",
"0xa8ef95dbd3db55911d3307930a84b27d6e969526",
# "0x4129c62faf652fea61375dcd9ca8ce24b2bb8b95",
"0x32885a6adac4375858E6edC092EfDDb0Ef46484C",
]
MAX_FILLS_TO_DISPLAY = 10
LOGS_DIR = "_logs"
recent_fills = {}
_lines_printed = 0
TABLE_HEADER = f"{'Time (UTC)':<10} | {'Coin':<6} | {'Side':<5} | {'Size':>15} | {'Price':>15} | {'Value (USD)':>20}"
TABLE_WIDTH = len(TABLE_HEADER)
def log_fill_to_csv(address: str, fill_data: dict):
"""Appends a single fill record to the CSV file for a specific address."""
log_file_path = os.path.join(LOGS_DIR, f"fills_{address}.csv")
file_exists = os.path.exists(log_file_path)
# The CSV will store a flattened version of the decoded fill
csv_row = {
'time_utc': fill_data['time'].isoformat(),
'coin': fill_data['coin'],
'side': fill_data['side'],
'price': fill_data['price'],
'size': fill_data['size'],
'value_usd': fill_data['value']
}
try:
with open(log_file_path, 'a', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=csv_row.keys())
if not file_exists:
writer.writeheader()
writer.writerow(csv_row)
except IOError as e:
logging.error(f"Failed to write to CSV log for {address}: {e}")
def on_message(message):
"""
Callback function to process incoming userEvents from the WebSocket.
"""
try:
logging.debug(f"Received message: {message}")
channel = message.get("channel")
if channel in ("user", "userFills"):
data = message.get("data")
if not data:
return
user_address = data.get("user", "").lower()
fills = data.get("fills", [])
if user_address in recent_fills and fills:
logging.info(f"Fill detected for user: {user_address}")
for fill_data in fills:
decoded_fill = {
"time": datetime.fromtimestamp(fill_data['time'] / 1000, tz=timezone.utc),
"coin": fill_data['coin'],
"side": "BUY" if fill_data['side'] == "B" else "SELL",
"price": float(fill_data['px']),
"size": float(fill_data['sz']),
"value": float(fill_data['px']) * float(fill_data['sz']),
}
recent_fills[user_address].append(decoded_fill)
# --- ADDED: Log every fill to its CSV file ---
log_fill_to_csv(user_address, decoded_fill)
except (KeyError, TypeError, ValueError) as e:
logging.error(f"Error processing message: {e} | Data: {message}")
def build_fills_table(address: str, fills: deque) -> list:
"""Builds the formatted lines for a single address's fills table."""
lines = []
short_address = f"{address[:6]}...{address[-4:]}"
lines.append(f"--- Fills for {short_address} ---")
lines.append(TABLE_HEADER)
lines.append("-" * TABLE_WIDTH)
for fill in list(fills):
lines.append(
f"{fill['time'].strftime('%H:%M:%S'):<10} | "
f"{fill['coin']:<6} | "
f"{fill['side']:<5} | "
f"{fill['size']:>15.4f} | "
f"{fill['price']:>15,.2f} | "
f"${fill['value']:>18,.2f}"
)
padding_needed = MAX_FILLS_TO_DISPLAY - len(fills)
for _ in range(padding_needed):
lines.append("")
return lines
def display_dashboard():
"""
Clears the screen and prints a two-column layout of recent fills tables.
"""
global _lines_printed
if _lines_printed > 0:
print(f"\x1b[{_lines_printed}A", end="")
output_lines = ["--- Live Address Fill Monitor ---", ""]
addresses_to_display = list(recent_fills.keys())
num_addresses = len(addresses_to_display)
mid_point = (num_addresses + 1) // 2
left_column_addresses = addresses_to_display[:mid_point]
right_column_addresses = addresses_to_display[mid_point:]
separator = " | "
for i in range(mid_point):
left_address = left_column_addresses[i]
left_table_lines = build_fills_table(left_address, recent_fills[left_address])
right_table_lines = []
if i < len(right_column_addresses):
right_address = right_column_addresses[i]
right_table_lines = build_fills_table(right_address, recent_fills[right_address])
table_height = 3 + MAX_FILLS_TO_DISPLAY
for j in range(table_height):
left_part = left_table_lines[j] if j < len(left_table_lines) else ""
right_part = right_table_lines[j] if j < len(right_table_lines) else ""
output_lines.append(f"{left_part:<{TABLE_WIDTH}}{separator}{right_part}")
output_lines.append("")
final_output = "\n".join(output_lines) + "\n\x1b[J"
print(final_output, end="")
_lines_printed = len(output_lines)
sys.stdout.flush()
def main():
"""
Main function to set up the WebSocket and run the display loop.
"""
global recent_fills
parser = argparse.ArgumentParser(description="Monitor live fills for specific wallet addresses on Hyperliquid.")
parser.add_argument(
"--addresses",
nargs='+',
default=DEFAULT_ADDRESSES_TO_WATCH,
help="A space-separated list of Ethereum addresses to monitor."
)
parser.add_argument(
"--log-level",
default="normal",
choices=['off', 'normal', 'debug'],
help="Set the logging level for the script."
)
args = parser.parse_args()
setup_logging(args.log_level, 'AddressMonitor')
# --- ADDED: Ensure the logs directory exists ---
if not os.path.exists(LOGS_DIR):
os.makedirs(LOGS_DIR)
addresses_to_watch = []
for addr in args.addresses:
clean_addr = addr.strip().lower()
if len(clean_addr) == 42 and clean_addr.startswith('0x'):
addresses_to_watch.append(clean_addr)
else:
logging.warning(f"Invalid or malformed address provided: '{addr}'. Skipping.")
recent_fills = {addr: deque(maxlen=MAX_FILLS_TO_DISPLAY) for addr in addresses_to_watch}
if not addresses_to_watch:
print("No valid addresses configured to watch. Exiting.", file=sys.stderr)
return
info = Info(constants.MAINNET_API_URL, skip_ws=False)
for addr in addresses_to_watch:
try:
info.subscribe({"type": "userFills", "user": addr}, on_message)
logging.debug(f"Queued subscribe for userFills: {addr}")
time.sleep(0.02)
except Exception as e:
logging.error(f"Failed to subscribe for {addr}: {e}")
logging.info(f"Subscribed to userFills for {len(addresses_to_watch)} addresses")
print("\nDisplaying live fill data... Press Ctrl+C to stop.")
try:
while True:
display_dashboard()
time.sleep(0.2)
except KeyboardInterrupt:
print("\nStopping WebSocket listener...")
info.ws_manager.stop()
print("Listener stopped.")
if __name__ == "__main__":
main()

View File

@ -1,368 +0,0 @@
import argparse
import logging
import os
import sys
import sqlite3
import pandas as pd
import json
from datetime import datetime, timedelta
import itertools
import multiprocessing
from functools import partial
import time
import importlib
import signal
from logging_utils import setup_logging
def _run_trade_simulation(df: pd.DataFrame, capital: float, size_pct: float, leverage_long: int, leverage_short: int, taker_fee_pct: float, maker_fee_pct: float) -> tuple[float, list]:
"""
Simulates a trading strategy with portfolio management, including capital,
position sizing, leverage, and fees.
"""
df.dropna(inplace=True)
if df.empty: return capital, []
df['position_change'] = df['signal'].diff()
trades = []
entry_price = 0
asset_size = 0
current_position = 0 # 0=flat, 1=long, -1=short
equity = capital
for i, row in df.iterrows():
# --- Close Positions ---
if (current_position == 1 and row['signal'] != 1) or \
(current_position == -1 and row['signal'] != -1):
exit_value = asset_size * row['close']
fee = exit_value * (taker_fee_pct / 100)
if current_position == 1: # Closing a long
pnl_usd = (row['close'] - entry_price) * asset_size
equity += pnl_usd - fee
trades.append({'pnl_usd': pnl_usd, 'pnl_pct': (row['close'] - entry_price) / entry_price, 'type': 'long'})
elif current_position == -1: # Closing a short
pnl_usd = (entry_price - row['close']) * asset_size
equity += pnl_usd - fee
trades.append({'pnl_usd': pnl_usd, 'pnl_pct': (entry_price - row['close']) / entry_price, 'type': 'short'})
entry_price = 0
asset_size = 0
current_position = 0
# --- Open New Positions ---
if current_position == 0:
if row['signal'] == 1: # Open Long
margin_to_use = equity * (size_pct / 100)
trade_value = margin_to_use * leverage_long
asset_size = trade_value / row['close']
fee = trade_value * (taker_fee_pct / 100)
equity -= fee
entry_price = row['close']
current_position = 1
elif row['signal'] == -1: # Open Short
margin_to_use = equity * (size_pct / 100)
trade_value = margin_to_use * leverage_short
asset_size = trade_value / row['close']
fee = trade_value * (taker_fee_pct / 100)
equity -= fee
entry_price = row['close']
current_position = -1
return equity, trades
def simulation_worker(params: dict, db_path: str, coin: str, timeframe: str, start_date: str, end_date: str, strategy_class, sim_params: dict) -> tuple[dict, float, list]:
"""
Worker function that loads data, runs the full simulation, and returns results.
"""
df = pd.DataFrame()
try:
with sqlite3.connect(db_path) as conn:
query = f'SELECT datetime_utc, open, high, low, close FROM "{coin}_{timeframe}" WHERE datetime_utc >= ? AND datetime_utc <= ? ORDER BY datetime_utc'
df = pd.read_sql(query, conn, params=(start_date, end_date), parse_dates=['datetime_utc'])
if not df.empty:
df.set_index('datetime_utc', inplace=True)
except Exception as e:
print(f"Worker error loading data for params {params}: {e}")
return (params, sim_params['capital'], [])
if df.empty:
return (params, sim_params['capital'], [])
strategy_instance = strategy_class(params)
df_with_signals = strategy_instance.calculate_signals(df)
final_equity, trades = _run_trade_simulation(df_with_signals, **sim_params)
return (params, final_equity, trades)
def init_worker():
signal.signal(signal.SIGINT, signal.SIG_IGN)
class Backtester:
def __init__(self, log_level: str, strategy_name_to_test: str, start_date: str, sim_params: dict):
setup_logging(log_level, 'Backtester')
self.db_path = os.path.join("_data", "market_data.db")
self.simulation_params = sim_params
self.backtest_config = self._load_backtest_config(strategy_name_to_test)
# ... (rest of __init__ is unchanged)
self.strategy_name = self.backtest_config.get('strategy_name')
self.strategy_config = self._load_strategy_config()
self.params = self.strategy_config.get('parameters', {})
self.coin = self.params.get('coin')
self.timeframe = self.params.get('timeframe')
self.pool = None
self.full_history_start_date = start_date
try:
module_path, class_name = self.backtest_config['script'].rsplit('.', 1)
module = importlib.import_module(module_path)
self.strategy_class = getattr(module, class_name)
logging.info(f"Successfully loaded strategy class '{class_name}'.")
except (ImportError, AttributeError, KeyError) as e:
logging.error(f"Could not load strategy script '{self.backtest_config.get('script')}': {e}")
sys.exit(1)
def _load_backtest_config(self, name_to_test: str):
# ... (unchanged)
config_path = os.path.join("_data", "backtesting_conf.json")
try:
with open(config_path, 'r') as f: return json.load(f).get(name_to_test)
except (FileNotFoundError, json.JSONDecodeError) as e:
logging.error(f"Could not load backtesting configuration: {e}")
return None
def _load_strategy_config(self):
# ... (unchanged)
config_path = os.path.join("_data", "strategies.json")
try:
with open(config_path, 'r') as f: return json.load(f).get(self.strategy_name)
except (FileNotFoundError, json.JSONDecodeError) as e:
logging.error(f"Could not load strategy configuration: {e}")
return None
def run_walk_forward_optimization(self, optimization_weeks: int, testing_weeks: int, step_weeks: int):
# ... (unchanged, will now use the new simulation logic via the worker)
full_df = self.load_data(self.full_history_start_date, datetime.now().strftime("%Y-%m-%d"))
if full_df.empty: return
optimization_delta = timedelta(weeks=optimization_weeks)
testing_delta = timedelta(weeks=testing_weeks)
step_delta = timedelta(weeks=step_weeks)
all_out_of_sample_trades = []
all_period_summaries = []
current_date = full_df.index[0]
end_date = full_df.index[-1]
period_num = 1
while current_date + optimization_delta + testing_delta <= end_date:
logging.info(f"\n--- Starting Walk-Forward Period {period_num} ---")
in_sample_start = current_date
in_sample_end = in_sample_start + optimization_delta
out_of_sample_end = in_sample_end + testing_delta
in_sample_df = full_df[in_sample_start:in_sample_end]
out_of_sample_df = full_df[in_sample_end:out_of_sample_end]
if in_sample_df.empty or out_of_sample_df.empty:
break
logging.info(f"In-Sample (Optimization): {in_sample_df.index[0].date()} to {in_sample_df.index[-1].date()}")
logging.info(f"Out-of-Sample (Testing): {out_of_sample_df.index[0].date()} to {out_of_sample_df.index[-1].date()}")
best_result = self._find_best_params(in_sample_df)
if not best_result:
all_period_summaries.append({"period": period_num, "params": "None Found"})
current_date += step_delta
period_num += 1
continue
print("\n--- [1] In-Sample Optimization Result ---")
print(f"Best Parameters Found: {best_result['params']}")
self._generate_report(best_result['final_equity'], best_result['trades_list'], "In-Sample Performance with Best Params")
logging.info(f"\n--- [2] Forward Testing on Out-of-Sample Data ---")
df_with_signals = self.strategy_class(best_result['params']).calculate_signals(out_of_sample_df.copy())
final_equity_oos, out_of_sample_trades = _run_trade_simulation(df_with_signals, **self.simulation_params)
all_out_of_sample_trades.extend(out_of_sample_trades)
oos_summary = self._generate_report(final_equity_oos, out_of_sample_trades, "Out-of-Sample Performance")
# Store the summary for the final table
summary_to_store = {"period": period_num, "params": best_result['params'], **oos_summary}
all_period_summaries.append(summary_to_store)
current_date += step_delta
period_num += 1
# ... (Final reports will be generated here, but need to adapt to equity tracking)
print("\n" + "="*50)
# self._generate_report(all_out_of_sample_trades, "FINAL AGGREGATE WALK-FORWARD PERFORMANCE")
print("="*50)
# --- ADDED: Final summary table of best parameters and performance per period ---
print("\n--- Summary of Best Parameters and Performance per Period ---")
header = f"{'#':<3} | {'Best Parameters':<30} | {'Trades':>8} | {'Longs':>6} | {'Shorts':>7} | {'Win %':>8} | {'L Win %':>9} | {'S Win %':>9} | {'Return %':>10} | {'Equity':>15}"
print(header)
print("-" * len(header))
for item in all_period_summaries:
params_str = str(item.get('params', 'N/A'))
trades = item.get('num_trades', 'N/A')
longs = item.get('num_longs', 'N/A')
shorts = item.get('num_shorts', 'N/A')
win_rate = f"{item.get('win_rate', 0):.2f}%" if 'win_rate' in item else 'N/A'
long_win_rate = f"{item.get('long_win_rate', 0):.2f}%" if 'long_win_rate' in item else 'N/A'
short_win_rate = f"{item.get('short_win_rate', 0):.2f}%" if 'short_win_rate' in item else 'N/A'
return_pct = f"{item.get('return_pct', 0):.2f}%" if 'return_pct' in item else 'N/A'
equity = f"${item.get('final_equity', 0):,.2f}" if 'final_equity' in item else 'N/A'
print(f"{item['period']:<3} | {params_str:<30} | {trades:>8} | {longs:>6} | {shorts:>7} | {win_rate:>8} | {long_win_rate:>9} | {short_win_rate:>9} | {return_pct:>10} | {equity:>15}")
def _find_best_params(self, df: pd.DataFrame) -> dict:
param_configs = self.backtest_config.get('optimization_params', {})
param_names = list(param_configs.keys())
param_ranges = [range(p['start'], p['end'] + 1, p['step']) for p in param_configs.values()]
all_combinations = list(itertools.product(*param_ranges))
param_dicts = [dict(zip(param_names, combo)) for combo in all_combinations]
logging.info(f"Optimizing on {len(all_combinations)} combinations...")
num_cores = 60
self.pool = multiprocessing.Pool(processes=num_cores, initializer=init_worker)
worker = partial(
simulation_worker,
db_path=self.db_path, coin=self.coin, timeframe=self.timeframe,
start_date=df.index[0].isoformat(), end_date=df.index[-1].isoformat(),
strategy_class=self.strategy_class,
sim_params=self.simulation_params
)
all_results = self.pool.map(worker, param_dicts)
self.pool.close()
self.pool.join()
self.pool = None
results = [{'params': params, 'final_equity': final_equity, 'trades_list': trades} for params, final_equity, trades in all_results if trades]
if not results: return None
return max(results, key=lambda x: x['final_equity'])
def load_data(self, start_date, end_date):
# ... (unchanged)
table_name = f"{self.coin}_{self.timeframe}"
logging.info(f"Loading full dataset for {table_name}...")
try:
with sqlite3.connect(self.db_path) as conn:
query = f'SELECT * FROM "{table_name}" WHERE datetime_utc >= ? AND datetime_utc <= ? ORDER BY datetime_utc'
df = pd.read_sql(query, conn, params=(start_date, end_date), parse_dates=['datetime_utc'])
if df.empty: return pd.DataFrame()
df.set_index('datetime_utc', inplace=True)
return df
except Exception as e:
logging.error(f"Failed to load data for backtest: {e}")
return pd.DataFrame()
def _generate_report(self, final_equity: float, trades: list, title: str) -> dict:
"""Calculates, prints, and returns a detailed performance report."""
print(f"\n--- {title} ---")
initial_capital = self.simulation_params['capital']
if not trades:
print("No trades were executed during this period.")
print(f"Final Equity: ${initial_capital:,.2f}")
return {"num_trades": 0, "num_longs": 0, "num_shorts": 0, "win_rate": 0, "long_win_rate": 0, "short_win_rate": 0, "return_pct": 0, "final_equity": initial_capital}
num_trades = len(trades)
long_trades = [t for t in trades if t.get('type') == 'long']
short_trades = [t for t in trades if t.get('type') == 'short']
pnls_pct = pd.Series([t['pnl_pct'] for t in trades])
wins = pnls_pct[pnls_pct > 0]
win_rate = (len(wins) / num_trades) * 100 if num_trades > 0 else 0
long_wins = len([t for t in long_trades if t['pnl_pct'] > 0])
short_wins = len([t for t in short_trades if t['pnl_pct'] > 0])
long_win_rate = (long_wins / len(long_trades)) * 100 if long_trades else 0
short_win_rate = (short_wins / len(short_trades)) * 100 if short_trades else 0
total_return_pct = ((final_equity - initial_capital) / initial_capital) * 100
print(f"Final Equity: ${final_equity:,.2f}")
print(f"Total Return: {total_return_pct:.2f}%")
print(f"Total Trades: {num_trades} (Longs: {len(long_trades)}, Shorts: {len(short_trades)})")
print(f"Win Rate (Overall): {win_rate:.2f}%")
print(f"Win Rate (Longs): {long_win_rate:.2f}%")
print(f"Win Rate (Shorts): {short_win_rate:.2f}%")
# Return a dictionary of the key metrics for the summary table
return {
"num_trades": num_trades,
"num_longs": len(long_trades),
"num_shorts": len(short_trades),
"win_rate": win_rate,
"long_win_rate": long_win_rate,
"short_win_rate": short_win_rate,
"return_pct": total_return_pct,
"final_equity": final_equity
}
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Run a Walk-Forward Optimization for a trading strategy.")
parser.add_argument("--strategy", required=True, help="The name of the backtest config to run.")
parser.add_argument("--start-date", default="2020-08-01", help="The overall start date for historical data.")
parser.add_argument("--optimization-weeks", type=int, default=4)
parser.add_argument("--testing-weeks", type=int, default=1)
parser.add_argument("--step-weeks", type=int, default=1)
parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug'])
parser.add_argument("--capital", type=float, default=1000)
parser.add_argument("--size-pct", type=float, default=50)
parser.add_argument("--leverage-long", type=int, default=3)
parser.add_argument("--leverage-short", type=int, default=2)
parser.add_argument("--taker-fee-pct", type=float, default=0.045)
parser.add_argument("--maker-fee-pct", type=float, default=0.015)
args = parser.parse_args()
sim_params = {
"capital": args.capital,
"size_pct": args.size_pct,
"leverage_long": args.leverage_long,
"leverage_short": args.leverage_short,
"taker_fee_pct": args.taker_fee_pct,
"maker_fee_pct": args.maker_fee_pct
}
backtester = Backtester(
log_level=args.log_level,
strategy_name_to_test=args.strategy,
start_date=args.start_date,
sim_params=sim_params
)
try:
backtester.run_walk_forward_optimization(
optimization_weeks=args.optimization_weeks,
testing_weeks=args.testing_weeks,
step_weeks=args.step_weeks
)
except KeyboardInterrupt:
logging.info("\nBacktest optimization cancelled by user.")
finally:
if backtester.pool:
logging.info("Terminating worker processes...")
backtester.pool.terminate()
backtester.pool.join()
logging.info("Worker processes terminated.")

View File

@ -1,169 +0,0 @@
from abc import ABC, abstractmethod
import pandas as pd
import json
import os
import logging
from datetime import datetime, timezone
import psycopg2
import multiprocessing
import time
from logging_utils import setup_logging
from hyperliquid.info import Info
from hyperliquid.utils import constants
class BaseStrategy(ABC):
"""
An abstract base class that defines the blueprint for all trading strategies.
It provides common functionality like loading data, saving status, and state management.
"""
def __init__(self, strategy_name: str, params: dict, trade_signal_queue: multiprocessing.Queue = None, shared_status: dict = None):
self.strategy_name = strategy_name
self.params = params
self.trade_signal_queue = trade_signal_queue
# Optional multiprocessing.Manager().dict() to hold live status (avoids file IO)
self.shared_status = shared_status
self.coin = params.get("coin", "N/A")
self.timeframe = params.get("timeframe", "N/A")
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"
self.last_signal_change_utc = None
self.signal_price = None
# Note: Logging is set up by the run_strategy function
def load_data(self) -> pd.DataFrame:
"""Loads historical data for the configured coin and 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:
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()
@abstractmethod
def calculate_signals(self, df: pd.DataFrame) -> pd.DataFrame:
"""The core logic of the strategy. Must be implemented by child classes."""
pass
def calculate_signals_and_state(self, df: pd.DataFrame) -> bool:
"""
A wrapper that calls the strategy's signal calculation, determines
the last signal change, and returns True if the signal has changed.
"""
df_with_signals = self.calculate_signals(df)
df_with_signals.dropna(inplace=True)
if df_with_signals.empty:
return False
df_with_signals['position_change'] = df_with_signals['signal'].diff()
last_signal_int = df_with_signals['signal'].iloc[-1]
new_signal_str = "HOLD"
if last_signal_int == 1: new_signal_str = "BUY"
elif last_signal_int == -1: new_signal_str = "SELL"
signal_changed = False
if self.current_signal == "INIT":
if new_signal_str == "BUY": self.current_signal = "INIT_BUY"
elif new_signal_str == "SELL": self.current_signal = "INIT_SELL"
else: self.current_signal = "HOLD"
signal_changed = True
elif new_signal_str != self.current_signal:
self.current_signal = new_signal_str
signal_changed = True
if signal_changed:
last_change_series = df_with_signals[df_with_signals['position_change'] != 0]
if not last_change_series.empty:
last_change_row = last_change_series.iloc[-1]
self.last_signal_change_utc = last_change_row.name.tz_localize('UTC').isoformat()
self.signal_price = last_change_row['close']
return signal_changed
def _save_status(self):
"""Saves the current strategy state to its JSON file."""
status = {
"strategy_name": self.strategy_name,
"current_signal": self.current_signal,
"last_signal_change_utc": self.last_signal_change_utc,
"signal_price": self.signal_price,
"last_checked_utc": datetime.now(timezone.utc).isoformat()
}
# If a shared status dict is provided (Manager.dict()), update it instead of writing files
try:
if self.shared_status is not None:
try:
# store the status under the strategy name for easy lookup
self.shared_status[self.strategy_name] = status
except Exception:
# Manager proxies may not accept nested mutable objects consistently; assign a copy
self.shared_status[self.strategy_name] = dict(status)
else:
with open(self.status_file_path, 'w', encoding='utf-8') as f:
json.dump(status, f, indent=4)
except IOError as e:
logging.error(f"Failed to write status file for {self.strategy_name}: {e}")
def run_polling_loop(self):
"""
The default execution loop for polling-based strategies (e.g., SMAs).
"""
while True:
df = self.load_data()
if df.empty:
logging.warning("No data loaded. Waiting 1 minute...")
time.sleep(60)
continue
signal_changed = self.calculate_signals_and_state(df.copy())
self._save_status()
if signal_changed or self.current_signal == "INIT_BUY" or self.current_signal == "INIT_SELL":
logging.warning(f"New signal detected: {self.current_signal}")
self.trade_signal_queue.put({
"strategy_name": self.strategy_name,
"signal": self.current_signal,
"coin": self.coin,
"signal_price": self.signal_price,
"config": {"agent": self.params.get("agent"), "parameters": self.params}
})
if self.current_signal == "INIT_BUY": self.current_signal = "BUY"
if self.current_signal == "INIT_SELL": self.current_signal = "SELL"
logging.info(f"Current Signal: {self.current_signal}")
time.sleep(60)
def run_event_loop(self):
"""
A placeholder for event-driven (WebSocket) strategies.
Child classes must override this.
"""
logging.error("run_event_loop() is not implemented for this strategy.")
time.sleep(3600) # Sleep for an hour to prevent rapid error loops
def on_fill_message(self, message):
"""
Placeholder for the WebSocket callback.
Child classes must override this.
"""
pass

View File

@ -1,31 +0,0 @@
import os
import sys
import time
import json
from datetime import datetime, timezone
from hyperliquid.info import Info
from hyperliquid.utils import constants
from collections import deque
def main():
address, info, _ = example_utils.setup(constants.MAINNET_API_URL)
# An example showing how to subscribe to the different subscription types and prints the returned messages
# Some subscriptions do not return snapshots, so you will not receive a message until something happens
info.subscribe({"type": "allMids"}, print)
info.subscribe({"type": "l2Book", "coin": "ETH"}, print)
info.subscribe({"type": "trades", "coin": "PURR/USDC"}, print)
info.subscribe({"type": "userEvents", "user": address}, print)
info.subscribe({"type": "userFills", "user": address}, print)
info.subscribe({"type": "candle", "coin": "ETH", "interval": "1m"}, print)
info.subscribe({"type": "orderUpdates", "user": address}, print)
info.subscribe({"type": "userFundings", "user": address}, print)
info.subscribe({"type": "userNonFundingLedgerUpdates", "user": address}, print)
info.subscribe({"type": "webData2", "user": address}, print)
info.subscribe({"type": "bbo", "coin": "ETH"}, print)
info.subscribe({"type": "activeAssetCtx", "coin": "BTC"}, print) # Perp
info.subscribe({"type": "activeAssetCtx", "coin": "@1"}, print) # Spot
info.subscribe({"type": "activeAssetData", "user": address, "coin": "BTC"}, print) # Perp only
if __name__ == "__main__":
main()

View File

@ -1,6 +0,0 @@
2025-12-11 14:29:08,607 - INFO - Strategy Initialized. Liquidity (L): 1236.4542
2025-12-11 14:29:09,125 - INFO - CLP Hedger initialized. Agent: 0xcB262CeAaE5D8A99b713f87a43Dd18E6Be892739. Coin: ETH (Decimals: 4)
2025-12-11 14:29:09,126 - INFO - Starting Hedge Monitor Loop. Interval: 30s
2025-12-11 14:29:09,126 - INFO - Hedging Range: 2844.11 - 3477.24 | Static Long: 0.4
2025-12-11 14:29:09,769 - INFO - Price: 3201.85 | Pool Delta: 0.883 | Tgt Short: 1.283 | Act Short: 0.000 | Diff: 1.283
2025-12-11 14:29:11,987 - ERROR - Order API Error: Order has invalid price.

View File

@ -1,18 +0,0 @@
[
{
"type": "MANUAL",
"token_id": 5147464,
"status": "OPEN",
"hedge_enabled": true,
"coin_symbol": "ETH",
"entry_price": 3332.66,
"range_lower": 2844.11,
"range_upper": 3477.24,
"target_value": 6938.95,
"amount0_initial": 0.45,
"amount1_initial": 5439.23,
"static_long": 0.0,
"timestamp_open": 1765575924,
"timestamp_close": null
}
]

View File

@ -1,214 +0,0 @@
import argparse
import json
import logging
import os
import sys
import time
from collections import deque
from datetime import datetime, timedelta
import csv
from hyperliquid.info import Info
from hyperliquid.utils import constants
from hyperliquid.utils.error import ClientError
# Assuming logging_utils.py is in the same directory
from logging_utils import setup_logging
class CandleFetcher:
"""
A class to fetch and manage historical candle data from Hyperliquid.
"""
def __init__(self, coins_to_fetch: list, interval: str, days_back: int):
self.info = Info(constants.MAINNET_API_URL, skip_ws=True)
self.coins = self._resolve_coins(coins_to_fetch)
self.interval = interval
self.days_back = days_back
self.data_folder = os.path.join("_data", "candles")
self.csv_headers = [
'datetime_utc', 'timestamp_ms', 'open', 'high', 'low', 'close', 'volume', 'number_of_trades'
]
self.header_mapping = {
't': 'timestamp_ms', 'o': 'open', 'h': 'high', 'l': 'low', 'c': 'close', 'v': 'volume', 'n': 'number_of_trades'
}
def _resolve_coins(self, coins_arg: list) -> list:
"""Determines the final list of coins to fetch."""
if coins_arg and "all" in [c.lower() for c in coins_arg]:
logging.info("Fetching data for all available coins.")
try:
with open("coin_precision.json", 'r') as f:
return list(json.load(f).keys())
except FileNotFoundError:
logging.error("'coin_precision.json' not found. Please run list_coins.py first.")
sys.exit(1)
else:
logging.info(f"Fetching data for specified coins: {coins_arg}")
return coins_arg
def run(self):
"""Starts the data fetching process for all configured coins."""
if not os.path.exists(self.data_folder):
os.makedirs(self.data_folder)
logging.info(f"Created data directory: '{self.data_folder}'")
for coin in self.coins:
logging.info(f"--- Starting process for {coin} ---")
self._update_data_for_coin(coin)
time.sleep(1) # Be polite to the API between processing different coins
def _get_start_time(self, file_path: str) -> (int, bool):
"""Checks for an existing file and returns the last timestamp, or a default start time."""
if os.path.exists(file_path):
try:
with open(file_path, 'r', newline='', encoding='utf-8') as f:
reader = csv.reader(f)
header = next(reader)
timestamp_index = header.index('timestamp_ms')
last_row = deque(reader, maxlen=1)
if last_row:
last_timestamp = int(last_row[0][timestamp_index])
logging.info(f"Existing file found. Resuming from timestamp: {last_timestamp}")
return last_timestamp, True
except (IOError, ValueError, StopIteration, IndexError) as e:
logging.warning(f"Could not read '{file_path}'. Re-fetching history. Error: {e}")
# If file doesn't exist or is invalid, fetch history
start_dt = datetime.now() - timedelta(days=self.days_back)
start_ms = int(start_dt.timestamp() * 1000)
logging.info(f"No valid data file. Fetching last {self.days_back} days.")
return start_ms, False
def _update_data_for_coin(self, coin: str):
"""Fetches and appends new candle data for a single coin."""
file_path = os.path.join(self.data_folder, f"{coin}_{self.interval}.csv")
start_time_ms, file_existed = self._get_start_time(file_path)
end_time_ms = int(time.time() * 1000)
if start_time_ms >= end_time_ms:
logging.warning(f"Start time ({datetime.fromtimestamp(start_time_ms/1000)}) is in the future. "
f"This can be caused by an incorrect system clock. No data will be fetched for {coin}.")
return
all_candles = self._fetch_candles_aggressively(coin, start_time_ms, end_time_ms)
if not all_candles:
logging.info(f"No new data found for {coin}.")
return
# --- FIX: Robust de-duplication and filtering ---
# This explicitly processes candles to ensure only new, unique ones are kept.
new_unique_candles = []
seen_timestamps = set()
# If updating an existing file, add the last known timestamp to the seen set
# to prevent re-adding the exact same candle.
if file_existed:
seen_timestamps.add(start_time_ms)
# Sort all fetched candles to process them chronologically
all_candles.sort(key=lambda c: c['t'])
for candle in all_candles:
timestamp = candle['t']
# Only process candles that are strictly newer than the last saved one
if timestamp > start_time_ms:
# Add the candle only if we haven't already added this timestamp
if timestamp not in seen_timestamps:
new_unique_candles.append(candle)
seen_timestamps.add(timestamp)
if new_unique_candles:
self._save_to_csv(new_unique_candles, file_path, file_existed)
else:
logging.info(f"No new candles to append for {coin}.")
def _fetch_candles_aggressively(self, coin, start_ms, end_ms):
"""
Uses a greedy, self-correcting loop to fetch data efficiently.
This is faster as it reduces the number of API calls.
"""
all_candles = []
current_start_time = start_ms
total_duration = end_ms - start_ms
while current_start_time < end_ms:
progress = ((current_start_time - start_ms) / total_duration) * 100 if total_duration > 0 else 100
current_time_str = datetime.fromtimestamp(current_start_time / 1000).strftime('%Y-%m-%d %H:%M:%S')
logging.info(f"Fetching {coin}: {progress:.2f}% complete. Current: {current_time_str}")
candle_batch = self._fetch_batch_with_retry(coin, current_start_time, end_ms)
if not candle_batch:
logging.info("No more candles returned from API. Fetch complete.")
break
all_candles.extend(candle_batch)
last_candle_timestamp = candle_batch[-1]["t"]
if last_candle_timestamp < current_start_time:
logging.warning("API returned older candles than requested. Breaking loop to prevent issues.")
break
current_start_time = last_candle_timestamp + 1
time.sleep(0.25) # Small delay to be polite
return all_candles
def _fetch_batch_with_retry(self, coin, start_ms, end_ms):
"""Performs a single API call with a retry mechanism."""
max_retries = 3
for attempt in range(max_retries):
try:
req = {"coin": coin, "interval": self.interval, "startTime": start_ms, "endTime": end_ms}
return self.info.post("/info", {"type": "candleSnapshot", "req": req})
except ClientError as e:
if e.status_code == 429 and attempt < max_retries - 1:
logging.warning("Rate limited. Retrying in 2 seconds...")
time.sleep(2)
else:
logging.error(f"API Error for {coin}: {e}. Skipping batch.")
return None
return None
def _save_to_csv(self, candles: list, file_path: str, is_append: bool):
"""Saves a list of candle data to a CSV file."""
processed_candles = []
for candle in candles:
new_candle = {self.header_mapping[k]: v for k, v in candle.items() if k in self.header_mapping}
new_candle['datetime_utc'] = datetime.fromtimestamp(candle['t'] / 1000).strftime('%Y-%m-%d %H:%M:%S')
processed_candles.append(new_candle)
write_mode = 'a' if is_append else 'w'
try:
with open(file_path, write_mode, newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=self.csv_headers)
if not is_append:
writer.writeheader()
writer.writerows(processed_candles)
logging.info(f"Successfully saved {len(processed_candles)} candles to '{file_path}'")
except IOError as e:
logging.error(f"Failed to write to file '{file_path}': {e}")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Fetch historical candle data from Hyperliquid.")
parser.add_argument(
"--coins",
nargs='+',
default=["BTC", "ETH"],
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).")
parser.add_argument("--days", type=int, default=7, help="Number of days of history to fetch for new coins.")
args = parser.parse_args()
setup_logging('normal', 'DataFetcher')
fetcher = CandleFetcher(coins_to_fetch=args.coins, interval=args.interval, days_back=args.days)
fetcher.run()

2
db.py
View File

@ -74,6 +74,8 @@ def upsert_candles(conn, table_name, records):
if not records: if not records:
return 0 return 0
records = list({r[1]: r for r in records}.values())
with conn.cursor() as cur: with conn.cursor() as cur:
execute_values( execute_values(
cur, cur,

View File

@ -1,5 +1,3 @@
version: "3.8"
services: services:
postgres: postgres:
image: postgres:15-alpine image: postgres:15-alpine
@ -8,29 +6,37 @@ services:
environment: environment:
POSTGRES_DB: hyper POSTGRES_DB: hyper
POSTGRES_USER: hyper POSTGRES_USER: hyper
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD} POSTGRES_PASSWORD: kaqpaaoi0
volumes: volumes:
- pg_data:/var/lib/postgresql/data - pg_data:/var/lib/postgresql/data
- ./postgres/postgresql.conf:/etc/postgresql/postgresql.conf - ./postgres/postgresql.conf:/etc/postgresql/postgresql.conf
command: postgres -c config_file=/etc/postgresql/postgresql.conf command: postgres -c config_file=/etc/postgresql/postgresql.conf
ports: ports:
- "5432:5432" - "5433:5432"
networks: networks:
- hyper_net - hyper_net
healthcheck:
test: ["CMD-SHELL", "pg_isready -U [secret] -d [secret]"]
interval: 10s
timeout: 5s
retries: 5
data-collector: data-collector:
build: . image: hyper-data-collector:latest
container_name: hyper_data container_name: hyper_data
restart: unless-stopped restart: unless-stopped
depends_on: depends_on:
- postgres postgres:
condition: service_healthy
env_file: env_file:
- .env.docker - .env.docker
environment:
- PYTHONPATH=/app
volumes: volumes:
- ./_data:/app/_data - ./_data:/app/_data
- ./_logs:/app/_logs - ./_logs:/app/_logs
- ./secrets:/app/secrets - ./secrets:/app/secrets
- /volume1/docker/hyper/backups:/backups - /volume2/docker/hyper/backups:/backups
networks: networks:
- hyper_net - hyper_net
@ -40,3 +46,6 @@ volumes:
networks: networks:
hyper_net: hyper_net:
driver: bridge driver: bridge
ipam:
config:
- subnet: 172.22.0.0/16

View File

@ -1,3 +1,4 @@
import os
import requests import requests
import json import json
import db import db

View File

@ -1,7 +1,7 @@
""" """
Indicator calculation module. Indicator calculation module.
Provides IndicatorCalculator for computing various financial indicators Provides IndicatorCalculator for computing various financial indicators
from SQLite candle data, including ratios, prices, moving averages, RSI, from PostgreSQL candle data, including ratios, prices, moving averages, RSI,
and custom functions. and custom functions.
""" """
@ -17,7 +17,7 @@ import numpy as np
class IndicatorCalculator: class IndicatorCalculator:
""" """
Computes indicator values from SQLite candle data. Computes indicator values from PostgreSQL candle data.
Supports ratio, price, spread, diff_pct, ma, rsi, and custom types. Supports ratio, price, spread, diff_pct, ma, rsi, and custom types.
""" """

View File

@ -2,7 +2,7 @@
Indicators Data Fetcher Indicators Data Fetcher
A standalone process that runs in a loop to compute financial indicators A standalone process that runs in a loop to compute financial indicators
(ratios, prices, MAs, RSI, custom) from SQLite candle data and save (ratios, prices, MAs, RSI, custom) from PostgreSQL candle data and save
the results to a JSON status file for the main dashboard to display. the results to a JSON status file for the main dashboard to display.
Follows the same pattern as dashboard_data_fetcher.py. Follows the same pattern as dashboard_data_fetcher.py.

137
list_latest_candles.py Normal file
View File

@ -0,0 +1,137 @@
import argparse
import json
import logging
import os
import sys
from datetime import datetime, timezone
from contextlib import closing
import psycopg2
from logging_utils import setup_logging
DEFAULT_DB_PATH = os.environ.get(
"PG_CONN_STR",
"postgresql://hyper:hyper@localhost:5432/hyper"
)
def load_coins():
"""Load the list of all coins from the local coin_precision.json file."""
coin_file = "_data/coin_precision.json"
try:
with open(coin_file, 'r') as f:
return list(json.load(f).keys())
except FileNotFoundError:
logging.error(f"'{coin_file}' not found. Please run list_coins.py first.")
sys.exit(1)
except (IOError, json.JSONDecodeError) as e:
logging.error(f"Failed to load or parse '{coin_file}': {e}")
sys.exit(1)
def get_latest_candle(conn, coin, interval="1m"):
"""
Query the database for the most recent candle for a given coin.
Returns a dict with keys: datetime_utc, timestamp_ms, open, high, low, close, volume, number_of_trades
or None if the table doesn't exist or has no rows.
"""
table_name = f"{coin.replace(':', '_')}_{interval}"
try:
with closing(conn.cursor()) as cur:
cur.execute(f'SELECT 1 FROM information_schema.tables WHERE table_name = %s', (table_name,))
if not cur.fetchone()[0]:
return None
cur.execute(
f'SELECT datetime_utc, timestamp_ms, open, high, low, close, volume, number_of_trades '
f'FROM "{table_name}" ORDER BY timestamp_ms DESC LIMIT 1'
)
row = cur.fetchone()
if row is None:
return None
return {
"datetime_utc": row[0],
"timestamp_ms": row[1],
"open": row[2],
"high": row[3],
"low": row[4],
"close": row[5],
"volume": row[6],
"number_of_trades": row[7],
}
except Exception as e:
logging.debug(f"Could not get latest candle for {coin} ({interval}): {e}")
return None
def list_latest_candles(coins, interval="1m", db_path=None):
"""
Fetch and display the newest candle for every coin in the list.
"""
if db_path is None:
db_path = DEFAULT_DB_PATH
conn = psycopg2.connect(db_path)
results = []
for coin in coins:
candle = get_latest_candle(conn, coin, interval)
if candle is not None:
results.append((coin, candle))
else:
results.append((coin, None))
conn.close()
print(f"\n--- Newest {interval} Candles for All Symbols ---")
print(f"Total symbols: {len(coins)} | Symbols with data: {sum(1 for _, c in results if c is not None)}")
print(f"Generated at: {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')}")
print("-" * 120)
print(f"{'Coin':<16} | {'Datetime (UTC)':<22} | {'Open':>12} | {'High':>12} | {'Low':>12} | {'Close':>12} | {'Volume':>12}")
print("-" * 120)
for coin, candle in results:
if candle is not None:
dt = candle["datetime_utc"].strftime('%Y-%m-%d %H:%M:%S') if candle["datetime_utc"] else "N/A"
o = f"{candle['open']:.4f}" if candle['open'] is not None else "N/A"
h = f"{candle['high']:.4f}" if candle['high'] is not None else "N/A"
l = f"{candle['low']:.4f}" if candle['low'] is not None else "N/A"
c = f"{candle['close']:.4f}" if candle['close'] is not None else "N/A"
v = f"{candle['volume']:.4f}" if candle['volume'] is not None else "N/A"
print(f"{coin:<16} | {dt:<22} | {o:>12} | {h:>12} | {l:>12} | {c:>12} | {v:>12}")
else:
print(f"{coin:<16} | {'(no data)':<22} | {'':>12} | {'':>12} | {'':>12} | {'':>12} | {'':>12}")
print("-" * 120)
print(f"Symbols without data: {sum(1 for _, c in results if c is None)}")
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="List the newest 1-minute candle for all symbols from the database."
)
parser.add_argument(
"--interval",
default="1m",
help="Candle interval to query (default: 1m)."
)
parser.add_argument(
"--db",
default=None,
help="PostgreSQL connection string (default: from PG_CONN_STR env or localhost)."
)
parser.add_argument(
"--log-level",
default="off",
choices=['off', 'normal', 'debug'],
help="Set the logging level."
)
args = parser.parse_args()
setup_logging(args.log_level, 'ListLatestCandles')
coins = load_coins()
list_latest_candles(coins, interval=args.interval, db_path=args.db)

View File

@ -1,150 +0,0 @@
from hyperliquid.info import Info
from hyperliquid.utils import constants
import time
import os
import sys
def get_asset_prices(asset_names=["BTC", "ETH", "SOL", "BNB", "FARTCOIN", "PUMP", "TRUMP", "ZEC"]):
"""
Connects to the Hyperliquid API to get the current mark price of specified assets.
Args:
asset_names (list): A list of asset names to retrieve prices for.
Returns:
list: A list of dictionaries, where each dictionary contains the name and mark price of an asset.
Returns an empty list if the API call fails or no assets are found.
"""
try:
info = Info(constants.MAINNET_API_URL, skip_ws=True)
meta, asset_contexts = info.meta_and_asset_ctxs()
universe = meta.get("universe", [])
asset_data = []
for name in asset_names:
try:
index = next(i for i, asset in enumerate(universe) if asset["name"] == name)
context = asset_contexts[index]
asset_data.append({
"name": name,
"mark_price": context.get("markPx")
})
except StopIteration:
print(f"Warning: Could not find asset '{name}' in the API response.")
return asset_data
except KeyError:
print("Error: A KeyError occurred. The structure of the API response may have changed.")
return []
except Exception as e:
print(f"An unexpected error occurred: {e}")
return []
def clear_console():
# Cross-platform clear screen
if os.name == 'nt':
os.system('cls')
else:
print('\033c', end='')
def display_prices_table(prices, previous_prices):
"""
Displays a list of asset prices in a formatted table with price change indicators.
Clears the console before displaying to keep the table in the same place.
Args:
prices (list): A list of asset data dictionaries from get_asset_prices.
previous_prices (dict): A dictionary of previous prices with asset names as keys.
"""
clear_console()
if not prices:
print("No price data to display.")
return
# Filter prices to only include assets in assets_to_track
tracked_assets = {asset['name'] for asset in assets_to_track}
prices = [asset for asset in prices if asset['name'] in tracked_assets]
# ANSI color codes
GREEN = '\033[92m'
RED = '\033[91m'
RESET = '\033[0m'
print(f"{'Asset':<12} | {'Mark Price':<20} | {'Change'}")
print("-" * 40)
for asset in prices:
current_price = float(asset['mark_price']) if asset['mark_price'] else 0
previous_price = previous_prices.get(asset['name'], 0)
indicator = " "
color = RESET
if previous_price and current_price > previous_price:
indicator = ""
color = GREEN
elif previous_price and current_price < previous_price:
indicator = ""
color = RED
# Use precision set in assets_to_track
precision = next((a['precision'] for a in assets_to_track if a['name'] == asset['name']), 2)
price_str = f"${current_price:,.{precision}f}" if current_price else "N/A"
print(f"{asset['name']:<12} | {color}{price_str:<20}{RESET} | {color}{indicator}{RESET}")
"""
Displays a list of asset prices in a formatted table with price change indicators.
Clears the console before displaying to keep the table in the same place.
Args:
prices (list): A list of asset data dictionaries from get_asset_prices.
previous_prices (dict): A dictionary of previous prices with asset names as keys.
"""
clear_console()
if not prices:
print("No price data to display.")
return
# ANSI color codes
GREEN = '\033[92m'
RED = '\033[91m'
RESET = '\033[0m'
print("\n")
print("-" * 38)
print(f"{'Asset':<8} | {'Mark Price':<15} | {'Change':<6} |")
print("-" * 38)
for asset in prices:
current_price = float(asset['mark_price']) if asset['mark_price'] else 0
previous_price = previous_prices.get(asset['name'], 0)
indicator = " "
color = RESET
if previous_price and current_price > previous_price:
indicator = ""
color = GREEN
elif previous_price and current_price < previous_price:
indicator = ""
color = RED
# Use precision set in assets_to_track
precision = next((a['precision'] for a in assets_to_track if a['name'] == asset['name']), 2)
price_str = f"${current_price:,.{precision}f}" if current_price else "N/A"
print(f"{asset['name']:<8} | {color}{price_str:<15}{RESET} | {color}{indicator:<4}{RESET} | ")
print("-" * 38)
if __name__ == "__main__":
assets_to_track = [
{"name": "BTC", "precision": 0}
]
previous_prices = {}
while True:
# Pass only the asset names to get_asset_prices
asset_names = [a["name"] for a in assets_to_track]
current_prices_data = get_asset_prices(asset_names)
display_prices_table(current_prices_data, previous_prices)
# Update previous_prices for the next iteration
for asset in current_prices_data:
if asset['mark_price']:
previous_prices[asset['name']] = float(asset['mark_price'])
time.sleep(1) # Add a delay to avoid overwhelming the API

View File

@ -1,175 +0,0 @@
import os
import sys
import time
import json
import argparse
from datetime import datetime, timezone
from hyperliquid.info import Info
from hyperliquid.utils import constants
from dotenv import load_dotenv
import logging
from logging_utils import setup_logging
# Load .env file
load_dotenv()
class PositionMonitor:
"""
A standalone, read-only dashboard for monitoring all open perpetuals
positions, spot balances, and their associated strategies.
"""
def __init__(self, log_level: str):
setup_logging(log_level, 'PositionMonitor')
self.wallet_address = os.environ.get("MAIN_WALLET_ADDRESS")
if not self.wallet_address:
logging.error("MAIN_WALLET_ADDRESS not set in .env file. Cannot proceed.")
sys.exit(1)
self.info = Info(constants.MAINNET_API_URL, skip_ws=True)
self.managed_positions_path = os.path.join("_data", "executor_managed_positions.json")
self._lines_printed = 0
logging.info(f"Monitoring vault address: {self.wallet_address}")
def load_managed_positions(self) -> dict:
"""Loads the state of which strategy manages which position."""
if os.path.exists(self.managed_positions_path):
try:
with open(self.managed_positions_path, 'r') as f:
# Create a reverse map: {coin: strategy_name}
data = json.load(f)
return {v['coin']: k for k, v in data.items()}
except (IOError, json.JSONDecodeError):
logging.warning("Could not read managed positions file.")
return {}
def run(self):
"""Main loop to continuously refresh the dashboard."""
try:
while True:
self.display_dashboard()
time.sleep(5) # Refresh every 5 seconds
except KeyboardInterrupt:
logging.info("Position monitor stopped.")
def display_dashboard(self):
"""Fetches all data and draws the dashboard without blinking."""
if self._lines_printed > 0:
print(f"\x1b[{self._lines_printed}A", end="")
output_lines = []
try:
perp_state = self.info.user_state(self.wallet_address)
spot_state = self.info.spot_user_state(self.wallet_address)
coin_to_strategy_map = self.load_managed_positions()
output_lines.append(f"--- Live Position Monitor for {self.wallet_address[:6]}...{self.wallet_address[-4:]} ---")
# --- 1. Perpetuals Account Summary ---
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
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 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:
self.build_spot_balances_table(spot_balances, output_lines)
# --- 3. Open Positions Table ---
output_lines.append("\n--- Open Perpetual Positions ---")
positions = perp_state.get('assetPositions', [])
open_positions = [p for p in positions if p.get('position') and float(p['position'].get('szi', 0)) != 0]
if not open_positions:
output_lines.append(" No open perpetual positions found.")
output_lines.append("") # Add a line for stable refresh
else:
self.build_positions_table(open_positions, coin_to_strategy_map, output_lines)
except Exception as e:
output_lines = [f"An error occurred: {e}"]
final_output = "\n".join(output_lines) + "\n\x1b[J" # \x1b[J clears to end of screen
print(final_output, end="")
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} |"
output_lines.append(header)
output_lines.append("-" * len(header))
for position in positions:
pos = position.get('position', {})
coin = pos.get('coin', 'Unknown')
size = float(pos.get('szi', 0))
entry_px = float(pos.get('entryPx', 0))
mark_px = float(pos.get('markPx', 0))
unrealized_pnl = float(pos.get('unrealizedPnl', 0))
# Get leverage
position_value = float(pos.get('positionValue', 0))
margin_used = float(pos.get('marginUsed', 0))
leverage = (position_value / margin_used) if margin_used > 0 else 0
side_text = "LONG" if size > 0 else "SHORT"
pnl_sign = "+" if unrealized_pnl >= 0 else ""
# Find the strategy that owns this coin
strategy_name = coin_to_strategy_map.get(coin, "Unmanaged")
# Format all values as strings
strategy_str = f"{strategy_name:<25}"
coin_str = f"{coin:<6}"
side_str = f"{side_text:<5}"
size_str = f"{size:>15.4f}"
entry_str = f"${entry_px:>11,.2f}"
mark_str = f"${mark_px:>11,.2f}"
pnl_str = f"{pnl_sign}${unrealized_pnl:>14,.2f}"
lev_str = f"{leverage:>9.1f}x"
output_lines.append(f"| {strategy_str} | {coin_str} | {side_str} | {size_str} | {entry_str} | {mark_str} | {pnl_str} | {lev_str} |")
output_lines.append("-" * len(header))
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Monitor a Hyperliquid wallet's positions in real-time.")
parser.add_argument(
"--log-level",
default="normal",
choices=['off', 'normal', 'debug'],
help="Set the logging level for the script."
)
args = parser.parse_args()
monitor = PositionMonitor(log_level=args.log_level)
monitor.run()

View File

@ -29,7 +29,6 @@ msgpack==1.1.2
multidict==6.7.0 multidict==6.7.0
numpy==2.0.2 numpy==2.0.2
pandas==2.3.3 pandas==2.3.3
parsimonious==0.10.0
propcache==0.4.1 propcache==0.4.1
pycares==4.11.0 pycares==4.11.0
pycparser==2.23 pycparser==2.23
@ -51,6 +50,6 @@ typing_extensions==4.15.0
tzdata==2025.2 tzdata==2025.2
urllib3==1.26.20 urllib3==1.26.20
websocket-client==1.9.0 websocket-client==1.9.0
web3~=6.0.0 # This means >=6.0.0 and <7.0.0 web3>=6.0.0,<7.0.0
yarl==1.22.0 yarl==1.22.0
psycopg2-binary==2.9.9 psycopg2-binary==2.9.9

View File

@ -2,17 +2,20 @@ import argparse
import logging import logging
import os import os
import sys import sys
import warnings
import db import db
import pandas as pd import pandas as pd
import json import json
from datetime import datetime, timezone, timedelta from datetime import datetime, timezone, timedelta
warnings.filterwarnings("ignore", message="pandas only supports SQLAlchemy")
# Assuming logging_utils.py is in the same directory # Assuming logging_utils.py is in the same directory
from logging_utils import setup_logging from logging_utils import setup_logging
class Resampler: class Resampler:
""" """
Reads new 1-minute candle data from the SQLite database, resamples it to Reads new 1-minute candle data from the PostgreSQL database, resamples it to
various timeframes, and upserts the new candles to the corresponding tables, various timeframes, and upserts the new candles to the corresponding tables,
preventing data duplication. preventing data duplication.
""" """
@ -79,13 +82,13 @@ class Resampler:
logging.debug(f"Processing {len(self.coins_to_process)} coins...") logging.debug(f"Processing {len(self.coins_to_process)} coins...")
for coin in self.coins_to_process: for coin in self.coins_to_process:
logging.debug(f"--- Processing {coin} ---") logging.info(f"--- Processing {coin} ---")
try: try:
for tf_name, tf_code in self.timeframes.items(): for tf_name, tf_code in self.timeframes.items():
target_table_name = db.sanitize_table_name(coin, tf_name) target_table_name = db.sanitize_table_name(coin, tf_name)
source_table_name = db.sanitize_table_name(coin, "1m") source_table_name = db.sanitize_table_name(coin, "1m")
logging.debug(f" Updating {tf_name} table...") logging.info(f" Resampling {coin} -> {tf_name}")
last_timestamp_ms = self._get_last_timestamp(conn, target_table_name) last_timestamp_ms = self._get_last_timestamp(conn, target_table_name)
@ -119,8 +122,8 @@ class Resampler:
records_to_upsert.append(( records_to_upsert.append((
index.strftime('%Y-%m-%d %H:%M:%S'), index.strftime('%Y-%m-%d %H:%M:%S'),
int(index.timestamp() * 1000), # Generate timestamp_ms int(index.timestamp() * 1000), # Generate timestamp_ms
row['open'], row['high'], row['low'], row['close'], float(row['open']), float(row['high']), float(row['low']), float(row['close']),
row['volume'], row['number_of_trades'] float(row['volume']), int(row['number_of_trades'])
)) ))
db.upsert_candles(conn, target_table_name, records_to_upsert) db.upsert_candles(conn, target_table_name, records_to_upsert)
@ -225,7 +228,7 @@ def parse_timeframes(tf_strings: list) -> dict:
if __name__ == "__main__": if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Resample 1-minute candle data from SQLite to other timeframes.") parser = argparse.ArgumentParser(description="Resample 1-minute candle data from PostgreSQL to other timeframes.")
parser.add_argument("--coins", nargs='+', required=True, help="List of coins to process.") 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("--timeframes", nargs='+', required=True, help="List of timeframes to generate.")
parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug']) parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug'])

View File

@ -8,7 +8,56 @@ The following sections detail recommendations for improving configuration manage
--- ---
## Proposed Code Changes ## Cleanup Status
The following cleanup actions have been completed:
### Deleted (Obsolete / Old Versions):
- `data_fetcher_old.py`
- `market_old.py`
- `base_strategy.py` (root; `strategies/base_strategy.py` is used)
- `strategy_sma_cross.py` (standalone old version; `strategies/ma_cross_strategy.py` is used)
### Deleted (Old Architecture Remnants):
- `address_monitor.py`
- `position_monitor.py`
- `trade_log.py`
- `wallet_data.py`
- `whale_tracker.py`
### Deleted (Zero-byte Docker Artifacts):
- `1a749d1ce7c2`, `37e7cf58e0c3`, `466c0182639b`, `65740cddd0af`, `6d00d75e1dce`, `851f9bf4c3cc`, `9dd58c972c63`, `d1611986dd76`, `d22d67dc5558`
- `Running`, `Using`
### Deleted (Runtime Artifacts):
- `clp_hedger.log`
- `clp_hedger/hedge_status.json`
### Moved to `scripts/`:
- `!migrate_to_sqlite.py``scripts/migrate_to_sqlite.py`
- `import_csv.py``scripts/import_csv.py`
- `del_market_cap_tables.py``scripts/del_market_cap_tables.py`
- `fix_timestamps.py``scripts/fix_timestamps.py`
- `list_coins.py``scripts/list_coins.py`
- `create_agent.py``scripts/create_agent.py`
- `check_wtioil.py``scripts/check_wtioil.py`
### Moved to `.temp/`:
- `strategy_template.py``.temp/strategy_template.py`
- `basic_ws.py``.temp/basic_ws.py`
- `backtester.py``.temp/backtester.py`
### `.gitignore` Updated:
- Added entries for `clp_hedger.log`, `clp_hedger/hedge_status.json`, `Using`, `Running`, and Docker layer hash files (`/[0-9a-f]{12}`)
### Example Config Files Created:
- `_data/strategies.json.example`
- `_data/backtesting_conf.json.example`
- `_data/coin_precision.json.example`
---
## Remaining Proposed Code Changes
### 1. Centralize Configuration ### 1. Centralize Configuration
@ -29,9 +78,8 @@ The following sections detail recommendations for improving configuration manage
### 3. Improve Project Structure ### 3. Improve Project Structure
- **Issue:** The root directory is cluttered with numerous Python scripts, making it difficult to distinguish between core application files, utility scripts, and old/example files. - **Issue:** The root directory is still somewhat cluttered with Python scripts.
- **Proposal:** - **Proposal:**
- Create a `scripts/` directory and move all one-off utility and maintenance scripts into it.
- Consider creating a `src/` or `app/` directory to house the core application source code (`main_app.py`, `trade_executor.py`, etc.), separating it clearly from configuration, data, and documentation. - Consider creating a `src/` or `app/` directory to house the core application source code (`main_app.py`, `trade_executor.py`, etc.), separating it clearly from configuration, data, and documentation.
- **Benefit:** A cleaner, more organized project structure that is easier for new developers to understand. - **Benefit:** A cleaner, more organized project structure that is easier for new developers to understand.
@ -41,39 +89,3 @@ The following sections detail recommendations for improving configuration manage
- **Proposal:** - **Proposal:**
- Explicitly define the agent names and their corresponding environment variable keys in the proposed `_data/config.json` file. The `trade_executor` would then load only the agents specified in the configuration. - Explicitly define the agent names and their corresponding environment variable keys in the proposed `_data/config.json` file. The `trade_executor` would then load only the agents specified in the configuration.
- **Benefit:** Makes agent configuration more explicit and less prone to errors from stray environment variables. - **Benefit:** Makes agent configuration more explicit and less prone to errors from stray environment variables.
---
## Identified Unused/Utility Files
The following files were identified as likely being unused by the core application, being obsolete, or serving as one-off utilities. It is recommended to **move them to a `scripts/` directory** or **delete them** if they are obsolete.
### Obsolete / Old Versions:
- `data_fetcher_old.py`
- `market_old.py`
- `base_strategy.py` (The one in the root directory; the one in `strategies/` is used).
### One-Off Utility Scripts (Recommend moving to `scripts/`):
- `!migrate_to_sqlite.py`
- `import_csv.py`
- `del_market_cap_tables.py`
- `fix_timestamps.py`
- `list_coins.py`
- `create_agent.py`
### Examples / Unused Code:
- `basic_ws.py` (Appears to be an example file).
- `backtester.py`
- `strategy_sma_cross.py` (A strategy file in the root, not in the `strategies` folder).
- `strategy_template.py`
### Standalone / Potentially Unused Core Files:
The following files seem to have their logic already integrated into the main multi-process application. They might be remnants of a previous architecture and may not be needed as standalone scripts.
- `address_monitor.py`
- `position_monitor.py`
- `trade_log.py`
- `wallet_data.py`
- `whale_tracker.py`
### Data / Log Files (Recommend archiving or deleting):
- `hyperliquid_wallet_data_*.json` (These appear to be backups or logs).

View File

@ -11,7 +11,7 @@ from logging_utils import setup_logging
class CsvImporter: class CsvImporter:
""" """
Imports historical candle data from a large CSV file into the SQLite database, Imports historical candle data from a large CSV file into the PostgreSQL database,
intelligently adding only the missing data. intelligently adding only the missing data.
""" """
@ -139,7 +139,7 @@ class CsvImporter:
if __name__ == "__main__": if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Import historical CSV data into the SQLite database.") parser = argparse.ArgumentParser(description="Import historical CSV data into the PostgreSQL database.")
parser.add_argument("--file", required=True, help="Path to the large CSV file to import.") parser.add_argument("--file", required=True, help="Path to the large CSV file to import.")
parser.add_argument("--coin", default="BTC", help="The coin symbol for this data (e.g., BTC).") parser.add_argument("--coin", default="BTC", help="The coin symbol for this data (e.g., BTC).")
parser.add_argument( parser.add_argument(

View File

@ -1,219 +0,0 @@
import argparse
import logging
import sys
import time
import pandas as pd
import sqlite3
import json
import os
from datetime import datetime, timezone, timedelta
from logging_utils import setup_logging
class SmaCrossStrategy:
"""
A flexible strategy that can operate in two modes:
1. Fast SMA / Slow SMA Crossover (if both 'fast' and 'slow' params are set)
2. Price / Single SMA Crossover (if only one 'fast' or 'slow' param is set)
"""
def __init__(self, strategy_name: str, params: dict, log_level: str):
self.strategy_name = strategy_name
self.params = params
self.coin = params.get("coin", "N/A")
self.timeframe = params.get("timeframe", "N/A")
# Load fast and slow SMA periods, defaulting to 0 if not present
self.fast_ma_period = params.get("fast", 0)
self.slow_ma_period = params.get("slow", 0)
self.db_path = os.path.join("_data", "market_data.db")
self.status_file_path = os.path.join("_data", f"strategy_status_{self.strategy_name}.json")
# Strategy state variables
self.current_signal = "INIT"
self.last_signal_change_utc = None
self.signal_price = None
self.fast_ma_value = None
self.slow_ma_value = None
setup_logging(log_level, f"Strategy-{self.strategy_name}")
logging.info(f"Initializing SMA Crossover strategy with parameters:")
for key, value in self.params.items():
logging.info(f" - {key}: {value}")
def load_data(self) -> pd.DataFrame:
"""Loads historical data, ensuring enough for the longest SMA calculation."""
table_name = f"{self.coin}_{self.timeframe}"
# Determine the longest period needed for calculations
longest_period = max(self.fast_ma_period or 0, self.slow_ma_period or 0)
if longest_period == 0:
logging.error("No valid SMA periods ('fast' or 'slow' > 0) are defined in parameters.")
return pd.DataFrame()
limit = longest_period + 50
try:
with sqlite3.connect(f"file:{self.db_path}?mode=ro", uri=True) as conn:
query = f'SELECT * FROM "{table_name}" ORDER BY datetime_utc DESC LIMIT {limit}'
df = pd.read_sql(query, conn)
if df.empty: return pd.DataFrame()
df['datetime_utc'] = pd.to_datetime(df['datetime_utc'])
df.set_index('datetime_utc', inplace=True)
df.sort_index(inplace=True)
return df
except Exception as e:
logging.error(f"Failed to load data from table '{table_name}': {e}")
return pd.DataFrame()
def _calculate_signals(self, data: pd.DataFrame):
"""
Analyzes historical data to find the last crossover event based on the
configured parameters (either dual or single SMA mode).
"""
# --- DUAL SMA CROSSOVER LOGIC ---
if self.fast_ma_period and self.slow_ma_period:
if len(data) < self.slow_ma_period + 1:
self.current_signal = "INSUFFICIENT DATA"
return
data['fast_sma'] = data['close'].rolling(window=self.fast_ma_period).mean()
data['slow_sma'] = data['close'].rolling(window=self.slow_ma_period).mean()
self.fast_ma_value = data['fast_sma'].iloc[-1]
self.slow_ma_value = data['slow_sma'].iloc[-1]
# Position is 1 for Golden Cross (fast > slow), -1 for Death Cross
data['position'] = 0
data.loc[data['fast_sma'] > data['slow_sma'], 'position'] = 1
data.loc[data['fast_sma'] < data['slow_sma'], 'position'] = -1
# --- SINGLE SMA PRICE CROSS LOGIC ---
else:
sma_period = self.fast_ma_period or self.slow_ma_period
if len(data) < sma_period + 1:
self.current_signal = "INSUFFICIENT DATA"
return
data['sma'] = data['close'].rolling(window=sma_period).mean()
self.slow_ma_value = data['sma'].iloc[-1] # Use slow_ma_value to store the single SMA
self.fast_ma_value = None # Ensure fast is None
# Position is 1 when price is above SMA, -1 when below
data['position'] = 0
data.loc[data['close'] > data['sma'], 'position'] = 1
data.loc[data['close'] < data['sma'], 'position'] = -1
# --- COMMON LOGIC for determining signal and last change ---
data['crossover'] = data['position'].diff()
last_position = data['position'].iloc[-1]
if last_position == 1: self.current_signal = "BUY"
elif last_position == -1: self.current_signal = "SELL"
else: self.current_signal = "HOLD"
last_cross_series = data[data['crossover'] != 0]
if not last_cross_series.empty:
last_cross_row = last_cross_series.iloc[-1]
self.last_signal_change_utc = last_cross_row.name.tz_localize('UTC').isoformat()
self.signal_price = last_cross_row['close']
if last_cross_row['position'] == 1: self.current_signal = "BUY"
elif last_cross_row['position'] == -1: self.current_signal = "SELL"
else:
self.last_signal_change_utc = data.index[0].tz_localize('UTC').isoformat()
self.signal_price = data['close'].iloc[0]
def _save_status(self):
"""Saves the current strategy state to its JSON file."""
status = {
"strategy_name": self.strategy_name,
"current_signal": self.current_signal,
"last_signal_change_utc": self.last_signal_change_utc,
"signal_price": self.signal_price,
"last_checked_utc": datetime.now(timezone.utc).isoformat()
}
try:
with open(self.status_file_path, 'w', encoding='utf-8') as f:
json.dump(status, f, indent=4)
except IOError as e:
logging.error(f"Failed to write status file: {e}")
def get_sleep_duration(self) -> int:
"""Calculates seconds to sleep until the next full candle closes."""
tf_value = int(''.join(filter(str.isdigit, self.timeframe)))
tf_unit = ''.join(filter(str.isalpha, self.timeframe))
if tf_unit == 'm': interval_seconds = tf_value * 60
elif tf_unit == 'h': interval_seconds = tf_value * 3600
elif tf_unit == 'd': interval_seconds = tf_value * 86400
else: return 60
now = datetime.now(timezone.utc)
timestamp = now.timestamp()
next_candle_ts = ((timestamp // interval_seconds) + 1) * interval_seconds
sleep_seconds = (next_candle_ts - timestamp) + 5
logging.info(f"Next candle closes at {datetime.fromtimestamp(next_candle_ts, tz=timezone.utc)}. "
f"Sleeping for {sleep_seconds:.2f} seconds.")
return sleep_seconds
def run_logic(self):
"""Main loop: loads data, calculates signals, saves status, and sleeps."""
logging.info(f"Starting logic loop for {self.coin} on {self.timeframe} timeframe.")
while True:
data = self.load_data()
if data.empty:
logging.warning("No data loaded. Waiting 1 minute before retrying...")
self.current_signal = "NO DATA"
self._save_status()
time.sleep(60)
continue
self._calculate_signals(data)
self._save_status()
last_close = data['close'].iloc[-1]
# --- Log based on which mode the strategy is running in ---
if self.fast_ma_period and self.slow_ma_period:
fast_ma_str = f"{self.fast_ma_value:.4f}" if self.fast_ma_value is not None else "N/A"
slow_ma_str = f"{self.slow_ma_value:.4f}" if self.slow_ma_value is not None else "N/A"
logging.info(
f"Signal: {self.current_signal} | Price: {last_close:.4f} | "
f"Fast SMA({self.fast_ma_period}): {fast_ma_str} | Slow SMA({self.slow_ma_period}): {slow_ma_str}"
)
else:
sma_period = self.fast_ma_period or self.slow_ma_period
sma_val_str = f"{self.slow_ma_value:.4f}" if self.slow_ma_value is not None else "N/A"
logging.info(
f"Signal: {self.current_signal} | Price: {last_close:.4f} | "
f"SMA({sma_period}): {sma_val_str}"
)
sleep_time = self.get_sleep_duration()
time.sleep(sleep_time)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Run an SMA Crossover trading strategy.")
parser.add_argument("--name", required=True, help="The name of the strategy instance from the config.")
parser.add_argument("--params", required=True, help="A JSON string of the strategy's parameters.")
parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug'])
args = parser.parse_args()
try:
strategy_params = json.loads(args.params)
strategy = SmaCrossStrategy(
strategy_name=args.name,
params=strategy_params,
log_level=args.log_level
)
strategy.run_logic()
except KeyboardInterrupt:
logging.info("Strategy process stopped.")
except Exception as e:
logging.error(f"A critical error occurred: {e}")
sys.exit(1)

View File

@ -1,186 +0,0 @@
import argparse
import logging
import sys
import time
import pandas as pd
import sqlite3
import json
import os
from datetime import datetime, timezone, timedelta
from logging_utils import setup_logging
class TradingStrategy:
"""
A template for a trading strategy that reads data from the SQLite database
and executes its logic in a loop, running once per candle.
"""
def __init__(self, strategy_name: str, params: dict, log_level: str):
self.strategy_name = strategy_name
self.params = params
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.status_file_path = os.path.join("_data", f"strategy_status_{self.strategy_name}.json")
# Strategy state variables
self.current_signal = "INIT"
self.last_signal_change_utc = None
self.signal_price = None
self.indicator_value = None
# Load strategy-specific parameters from config
self.rsi_period = params.get("rsi_period")
self.short_ma = params.get("short_ma")
self.long_ma = params.get("long_ma")
self.sma_period = params.get("sma_period")
setup_logging(log_level, f"Strategy-{self.strategy_name}")
logging.info(f"Initializing strategy with parameters: {self.params}")
def load_data(self) -> pd.DataFrame:
"""Loads historical data, ensuring enough for the longest indicator period."""
table_name = f"{self.coin}_{self.timeframe}"
limit = 500
# Determine required data limit based on the longest configured indicator
periods = [p for p in [self.sma_period, self.long_ma, self.rsi_period] if p is not None]
if periods:
limit = max(periods) + 50
try:
with sqlite3.connect(f"file:{self.db_path}?mode=ro", uri=True) as conn:
query = f'SELECT * FROM "{table_name}" ORDER BY datetime_utc DESC LIMIT {limit}'
df = pd.read_sql(query, conn)
if df.empty: return pd.DataFrame()
df['datetime_utc'] = pd.to_datetime(df['datetime_utc'])
df.set_index('datetime_utc', inplace=True)
df.sort_index(inplace=True)
return df
except Exception as e:
logging.error(f"Failed to load data from table '{table_name}': {e}")
return pd.DataFrame()
def _calculate_signals(self, data: pd.DataFrame):
"""
Analyzes historical data to find the last signal crossover event.
This method should be expanded to handle different strategy types.
"""
if self.sma_period:
if len(data) < self.sma_period + 1:
self.current_signal = "INSUFFICIENT DATA"
return
data['sma'] = data['close'].rolling(window=self.sma_period).mean()
self.indicator_value = data['sma'].iloc[-1]
data['position'] = 0
data.loc[data['close'] > data['sma'], 'position'] = 1
data.loc[data['close'] < data['sma'], 'position'] = -1
data['crossover'] = data['position'].diff()
last_position = data['position'].iloc[-1]
if last_position == 1: self.current_signal = "BUY"
elif last_position == -1: self.current_signal = "SELL"
else: self.current_signal = "HOLD"
last_cross_series = data[data['crossover'] != 0]
if not last_cross_series.empty:
last_cross_row = last_cross_series.iloc[-1]
self.last_signal_change_utc = last_cross_row.name.tz_localize('UTC').isoformat()
self.signal_price = last_cross_row['close']
if last_cross_row['position'] == 1: self.current_signal = "BUY"
elif last_cross_row['position'] == -1: self.current_signal = "SELL"
else:
self.last_signal_change_utc = data.index[0].tz_localize('UTC').isoformat()
self.signal_price = data['close'].iloc[0]
elif self.rsi_period:
logging.info(f"RSI logic not implemented for period {self.rsi_period}.")
self.current_signal = "NOT IMPLEMENTED"
elif self.short_ma and self.long_ma:
logging.info(f"MA Cross logic not implemented for {self.short_ma}/{self.long_ma}.")
self.current_signal = "NOT IMPLEMENTED"
def _save_status(self):
"""Saves the current strategy state to its JSON file."""
status = {
"strategy_name": self.strategy_name,
"current_signal": self.current_signal,
"last_signal_change_utc": self.last_signal_change_utc,
"signal_price": self.signal_price,
"last_checked_utc": datetime.now(timezone.utc).isoformat()
}
try:
with open(self.status_file_path, 'w', encoding='utf-8') as f:
json.dump(status, f, indent=4)
except IOError as e:
logging.error(f"Failed to write status file: {e}")
def get_sleep_duration(self) -> int:
"""Calculates seconds to sleep until the next full candle closes."""
if not self.timeframe: return 60
tf_value = int(''.join(filter(str.isdigit, self.timeframe)))
tf_unit = ''.join(filter(str.isalpha, self.timeframe))
if tf_unit == 'm': interval_seconds = tf_value * 60
elif tf_unit == 'h': interval_seconds = tf_value * 3600
elif tf_unit == 'd': interval_seconds = tf_value * 86400
else: return 60
now = datetime.now(timezone.utc)
timestamp = now.timestamp()
next_candle_ts = ((timestamp // interval_seconds) + 1) * interval_seconds
sleep_seconds = (next_candle_ts - timestamp) + 5
logging.info(f"Next candle closes at {datetime.fromtimestamp(next_candle_ts, tz=timezone.utc)}. "
f"Sleeping for {sleep_seconds:.2f} seconds.")
return sleep_seconds
def run_logic(self):
"""Main loop: loads data, calculates signals, saves status, and sleeps."""
logging.info(f"Starting main logic loop for {self.coin} on {self.timeframe} timeframe.")
while True:
data = self.load_data()
if data.empty:
logging.warning("No data loaded. Waiting 1 minute before retrying...")
self.current_signal = "NO DATA"
self._save_status()
time.sleep(60)
continue
self._calculate_signals(data)
self._save_status()
last_close = data['close'].iloc[-1]
indicator_val_str = f"{self.indicator_value:.4f}" if self.indicator_value is not None else "N/A"
logging.info(f"Signal: {self.current_signal} | Price: {last_close:.4f} | Indicator: {indicator_val_str}")
sleep_time = self.get_sleep_duration()
time.sleep(sleep_time)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Run a trading strategy.")
parser.add_argument("--name", required=True, help="The name of the strategy instance from the config.")
parser.add_argument("--params", required=True, help="A JSON string of the strategy's parameters.")
parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug'])
args = parser.parse_args()
try:
strategy_params = json.loads(args.params)
strategy = TradingStrategy(
strategy_name=args.name,
params=strategy_params,
log_level=args.log_level
)
strategy.run_logic()
except KeyboardInterrupt:
logging.info("Strategy process stopped.")
except Exception as e:
logging.error(f"A critical error occurred: {e}")
sys.exit(1)

View File

@ -1,55 +0,0 @@
import os
import csv
from datetime import datetime, timezone
import threading
# A lock to prevent race conditions when multiple strategies might log at once in the future
log_lock = threading.Lock()
def log_trade(strategy: str, coin: str, action: str, price: float, size: float, signal: str, pnl: float = 0.0):
"""
Appends a record of a trade action to a persistent CSV log file.
Args:
strategy (str): The name of the strategy that triggered the action.
coin (str): The coin being traded (e.g., 'BTC').
action (str): The action taken (e.g., 'OPEN_LONG', 'CLOSE_LONG').
price (float): The execution price of the trade.
size (float): The size of the trade.
signal (str): The signal that triggered the trade (e.g., 'BUY', 'SELL').
pnl (float, optional): The realized profit and loss for closing trades. Defaults to 0.0.
"""
log_dir = "_logs"
file_path = os.path.join(log_dir, "trade_history.csv")
# Ensure the logs directory exists
if not os.path.exists(log_dir):
os.makedirs(log_dir)
# Define the headers for the CSV file
headers = ["timestamp_utc", "strategy", "coin", "action", "price", "size", "signal", "pnl"]
# Check if the file needs a header
file_exists = os.path.isfile(file_path)
with log_lock:
try:
with open(file_path, 'a', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=headers)
if not file_exists:
writer.writeheader()
writer.writerow({
"timestamp_utc": datetime.now(timezone.utc).isoformat(),
"strategy": strategy,
"coin": coin,
"action": action,
"price": price,
"size": size,
"signal": signal,
"pnl": pnl
})
except IOError as e:
# If logging fails, print an error to the main console as a fallback.
print(f"CRITICAL: Failed to write to trade log file: {e}")

View File

@ -1,652 +0,0 @@
#!/usr/bin/env python3
"""
Hyperliquid Wallet Data Fetcher - FINAL Perfect Alignment
==========================================================
Complete Python script to pull all available data for a Hyperliquid wallet via API.
Requirements:
pip install hyperliquid-python-sdk
Usage:
python hyperliquid_wallet_data.py <wallet_address>
Example:
python hyperliquid_wallet_data.py 0xcd5051944f780a621ee62e39e493c489668acf4d
"""
import sys
import json
from datetime import datetime, timedelta
from typing import Optional, Dict, Any
from hyperliquid.info import Info
from hyperliquid.utils import constants
class HyperliquidWalletAnalyzer:
"""
Comprehensive wallet data analyzer for Hyperliquid exchange.
Fetches all available information about a specific wallet address.
"""
def __init__(self, wallet_address: str, use_testnet: bool = False):
"""
Initialize the analyzer with a wallet address.
Args:
wallet_address: Ethereum-style address (0x...)
use_testnet: If True, use testnet instead of mainnet
"""
self.wallet_address = wallet_address
api_url = constants.TESTNET_API_URL if use_testnet else constants.MAINNET_API_URL
# Initialize Info API (read-only, no private keys needed)
self.info = Info(api_url, skip_ws=True)
print(f"Initialized Hyperliquid API: {'Testnet' if use_testnet else 'Mainnet'}")
print(f"Target wallet: {wallet_address}\n")
def print_position_details(self, position: Dict[str, Any], index: int):
"""
Print detailed information about a single position.
Args:
position: Position data dictionary
index: Position number for display
"""
pos = position.get('position', {})
# Extract all position details
coin = pos.get('coin', 'Unknown')
size = float(pos.get('szi', 0))
entry_px = float(pos.get('entryPx', 0))
position_value = float(pos.get('positionValue', 0))
unrealized_pnl = float(pos.get('unrealizedPnl', 0))
return_on_equity = float(pos.get('returnOnEquity', 0))
# Leverage details
leverage = pos.get('leverage', {})
leverage_type = leverage.get('type', 'unknown') if isinstance(leverage, dict) else 'cross'
leverage_value = leverage.get('value', 0) if isinstance(leverage, dict) else 0
# Margin and liquidation
margin_used = float(pos.get('marginUsed', 0))
liquidation_px = pos.get('liquidationPx')
max_trade_szs = pos.get('maxTradeSzs', [0, 0])
# Cumulative funding
cumulative_funding = float(pos.get('cumFunding', {}).get('allTime', 0))
# Determine if long or short
side = "LONG 📈" if size > 0 else "SHORT 📉"
side_color = "🟢" if size > 0 else "🔴"
# PnL color
pnl_symbol = "🟢" if unrealized_pnl >= 0 else "🔴"
pnl_sign = "+" if unrealized_pnl >= 0 else ""
# ROE color
roe_symbol = "🟢" if return_on_equity >= 0 else "🔴"
roe_sign = "+" if return_on_equity >= 0 else ""
print(f"\n{'='*80}")
print(f"POSITION #{index}: {coin} {side} {side_color}")
print(f"{'='*80}")
print(f"\n📊 POSITION DETAILS:")
print(f" Size: {abs(size):.6f} {coin}")
print(f" Side: {side}")
print(f" Entry Price: ${entry_px:,.4f}")
print(f" Position Value: ${abs(position_value):,.2f}")
print(f"\n💰 PROFITABILITY:")
print(f" Unrealized PnL: {pnl_symbol} {pnl_sign}${unrealized_pnl:,.2f}")
print(f" Return on Equity: {roe_symbol} {roe_sign}{return_on_equity:.2%}")
print(f" Cumulative Funding: ${cumulative_funding:,.4f}")
print(f"\n⚙️ LEVERAGE & MARGIN:")
print(f" Leverage Type: {leverage_type.upper()}")
print(f" Leverage: {leverage_value}x")
print(f" Margin Used: ${margin_used:,.2f}")
print(f"\n⚠️ RISK MANAGEMENT:")
if liquidation_px:
liquidation_px_float = float(liquidation_px) if liquidation_px else 0
print(f" Liquidation Price: ${liquidation_px_float:,.4f}")
# Calculate distance to liquidation
if entry_px > 0 and liquidation_px_float > 0:
if size > 0: # Long position
distance = ((entry_px - liquidation_px_float) / entry_px) * 100
else: # Short position
distance = ((liquidation_px_float - entry_px) / entry_px) * 100
distance_symbol = "🟢" if abs(distance) > 20 else "🟡" if abs(distance) > 10 else "🔴"
print(f" Distance to Liq: {distance_symbol} {abs(distance):.2f}%")
else:
print(f" Liquidation Price: N/A (Cross margin)")
if max_trade_szs and len(max_trade_szs) == 2:
print(f" Max Long Trade: {max_trade_szs[0]}")
print(f" Max Short Trade: {max_trade_szs[1]}")
print(f"\n{'='*80}")
def get_user_state(self) -> Dict[str, Any]:
"""
Get complete user state including positions and margin summary.
Returns:
Dict containing:
- assetPositions: List of open perpetual positions
- marginSummary: Account value, margin used, withdrawable
- crossMarginSummary: Cross margin details
- withdrawable: Available balance to withdraw
"""
print("📊 Fetching User State (Perpetuals)...")
try:
data = self.info.user_state(self.wallet_address)
if data:
margin_summary = data.get('marginSummary', {})
positions = data.get('assetPositions', [])
account_value = float(margin_summary.get('accountValue', 0))
total_margin_used = float(margin_summary.get('totalMarginUsed', 0))
total_ntl_pos = float(margin_summary.get('totalNtlPos', 0))
total_raw_usd = float(margin_summary.get('totalRawUsd', 0))
withdrawable = float(data.get('withdrawable', 0))
print(f" ✓ Account Value: ${account_value:,.2f}")
print(f" ✓ Total Margin Used: ${total_margin_used:,.2f}")
print(f" ✓ Total Position Value: ${total_ntl_pos:,.2f}")
print(f" ✓ Withdrawable: ${withdrawable:,.2f}")
print(f" ✓ Open Positions: {len(positions)}")
# Calculate margin utilization
if account_value > 0:
margin_util = (total_margin_used / account_value) * 100
util_symbol = "🟢" if margin_util < 50 else "🟡" if margin_util < 75 else "🔴"
print(f" ✓ Margin Utilization: {util_symbol} {margin_util:.2f}%")
# Print detailed information for each position
if positions:
print(f"\n{'='*80}")
print(f"DETAILED POSITION BREAKDOWN ({len(positions)} positions)")
print(f"{'='*80}")
for idx, position in enumerate(positions, 1):
self.print_position_details(position, idx)
# Summary table with perfect alignment
self.print_positions_summary_table(positions)
else:
print(" ⚠ No perpetual positions found")
return data
except Exception as e:
print(f" ✗ Error: {e}")
return {}
def print_positions_summary_table(self, positions: list):
"""
Print a summary table of all positions with perfectly aligned columns.
NO emojis in data cells - keeps them simple text only for perfect alignment.
Args:
positions: List of position dictionaries
"""
print(f"\n{'='*130}")
print("POSITIONS SUMMARY TABLE")
print('='*130)
# Print header
print("| Asset | Side | Size | Entry Price | Position Value | Unrealized PnL | ROE | Leverage |")
print("|----------|-------|-------------------|-------------------|-------------------|-------------------|------------|------------|")
total_position_value = 0
total_pnl = 0
for position in positions:
pos = position.get('position', {})
coin = pos.get('coin', 'Unknown')
size = float(pos.get('szi', 0))
entry_px = float(pos.get('entryPx', 0))
position_value = float(pos.get('positionValue', 0))
unrealized_pnl = float(pos.get('unrealizedPnl', 0))
return_on_equity = float(pos.get('returnOnEquity', 0))
# Get leverage
leverage = pos.get('leverage', {})
leverage_value = leverage.get('value', 0) if isinstance(leverage, dict) else 0
leverage_type = leverage.get('type', 'cross') if isinstance(leverage, dict) else 'cross'
# Determine side - NO EMOJIS in data
side_text = "LONG" if size > 0 else "SHORT"
# Format PnL and ROE with signs
pnl_sign = "+" if unrealized_pnl >= 0 else ""
roe_sign = "+" if return_on_equity >= 0 else ""
# Accumulate totals
total_position_value += abs(position_value)
total_pnl += unrealized_pnl
# Format all values as strings with proper width
asset_str = f"{coin[:8]:<8}"
side_str = f"{side_text:<5}"
size_str = f"{abs(size):>17,.4f}"
entry_str = f"${entry_px:>16,.2f}"
value_str = f"${abs(position_value):>16,.2f}"
pnl_str = f"{pnl_sign}${unrealized_pnl:>15,.2f}"
roe_str = f"{roe_sign}{return_on_equity:>9.2%}"
lev_str = f"{leverage_value}x {leverage_type[:4]}"
# Print row with exact spacing
print(f"| {asset_str} | {side_str} | {size_str} | {entry_str} | {value_str} | {pnl_str} | {roe_str} | {lev_str:<10} |")
# Separator before totals
print("|==========|=======|===================|===================|===================|===================|============|============|")
# Total row
total_value_str = f"${total_position_value:>16,.2f}"
total_pnl_sign = "+" if total_pnl >= 0 else ""
total_pnl_str = f"{total_pnl_sign}${total_pnl:>15,.2f}"
print(f"| TOTAL | | | | {total_value_str} | {total_pnl_str} | | |")
print('='*130 + '\n')
def get_spot_state(self) -> Dict[str, Any]:
"""
Get spot trading state including token balances.
Returns:
Dict containing:
- balances: List of spot token holdings
"""
print("\n💰 Fetching Spot State...")
try:
data = self.info.spot_user_state(self.wallet_address)
if data and data.get('balances'):
print(f" ✓ Spot Holdings: {len(data['balances'])} tokens")
for balance in data['balances'][:5]: # Show first 5
print(f" - {balance.get('coin', 'Unknown')}: {balance.get('total', 0)}")
else:
print(" ⚠ No spot holdings found")
return data
except Exception as e:
print(f" ✗ Error: {e}")
return {}
def get_open_orders(self) -> list:
"""
Get all open orders for the user.
Returns:
List of open orders with details (price, size, side, etc.)
"""
print("\n📋 Fetching Open Orders...")
try:
data = self.info.open_orders(self.wallet_address)
if data:
print(f" ✓ Open Orders: {len(data)}")
for order in data[:3]: # Show first 3
coin = order.get('coin', 'Unknown')
side = order.get('side', 'Unknown')
size = order.get('sz', 0)
price = order.get('limitPx', 0)
print(f" - {coin} {side}: {size} @ ${price}")
else:
print(" ⚠ No open orders")
return data
except Exception as e:
print(f" ✗ Error: {e}")
return []
def get_user_fills(self, limit: int = 100) -> list:
"""
Get recent trade fills (executions).
Args:
limit: Maximum number of fills to retrieve (max 2000)
Returns:
List of fills with execution details, PnL, timestamps
"""
print(f"\n📈 Fetching Recent Fills (last {limit})...")
try:
data = self.info.user_fills(self.wallet_address)
if data:
fills = data[:limit]
print(f" ✓ Total Fills Retrieved: {len(fills)}")
# Show summary stats
total_pnl = sum(float(f.get('closedPnl', 0)) for f in fills if f.get('closedPnl'))
print(f" ✓ Total Closed PnL: ${total_pnl:.2f}")
# Show most recent
if fills:
recent = fills[0]
print(f" ✓ Most Recent: {recent.get('coin')} {recent.get('side')} {recent.get('sz')} @ ${recent.get('px')}")
else:
print(" ⚠ No fills found")
return data[:limit] if data else []
except Exception as e:
print(f" ✗ Error: {e}")
return []
def get_user_fills_by_time(self, start_time: Optional[int] = None,
end_time: Optional[int] = None) -> list:
"""
Get fills within a specific time range.
Args:
start_time: Start timestamp in milliseconds (default: 7 days ago)
end_time: End timestamp in milliseconds (default: now)
Returns:
List of fills within the time range
"""
if not start_time:
start_time = int((datetime.now() - timedelta(days=7)).timestamp() * 1000)
if not end_time:
end_time = int(datetime.now().timestamp() * 1000)
print(f"\n📅 Fetching Fills by Time Range...")
print(f" From: {datetime.fromtimestamp(start_time/1000)}")
print(f" To: {datetime.fromtimestamp(end_time/1000)}")
try:
data = self.info.user_fills_by_time(self.wallet_address, start_time, end_time)
if data:
print(f" ✓ Fills in Range: {len(data)}")
else:
print(" ⚠ No fills in this time range")
return data
except Exception as e:
print(f" ✗ Error: {e}")
return []
def get_user_fees(self) -> Dict[str, Any]:
"""
Get user's fee schedule and trading volume.
Returns:
Dict containing:
- feeSchedule: Fee rates by tier
- userCrossRate: User's current cross trading fee rate
- userAddRate: User's maker fee rate
- userWithdrawRate: Withdrawal fee rate
- dailyUserVlm: Daily trading volume
"""
print("\n💳 Fetching Fee Information...")
try:
data = self.info.user_fees(self.wallet_address)
if data:
print(f" ✓ Maker Fee: {data.get('userAddRate', 0)}%")
print(f" ✓ Taker Fee: {data.get('userCrossRate', 0)}%")
print(f" ✓ Daily Volume: ${data.get('dailyUserVlm', [0])[0] if data.get('dailyUserVlm') else 0}")
return data
except Exception as e:
print(f" ✗ Error: {e}")
return {}
def get_user_rate_limit(self) -> Dict[str, Any]:
"""
Get API rate limit information.
Returns:
Dict containing:
- cumVlm: Cumulative trading volume
- nRequestsUsed: Number of requests used
- nRequestsCap: Request capacity
"""
print("\n⏱️ Fetching Rate Limit Info...")
try:
data = self.info.user_rate_limit(self.wallet_address)
if data:
used = data.get('nRequestsUsed', 0)
cap = data.get('nRequestsCap', 0)
print(f" ✓ API Requests: {used}/{cap}")
print(f" ✓ Cumulative Volume: ${data.get('cumVlm', 0)}")
return data
except Exception as e:
print(f" ✗ Error: {e}")
return {}
def get_funding_history(self, coin: str, days: int = 7) -> list:
"""
Get funding rate history for a specific coin.
Args:
coin: Asset symbol (e.g., 'BTC', 'ETH')
days: Number of days of history (default: 7)
Returns:
List of funding rate entries
"""
end_time = int(datetime.now().timestamp() * 1000)
start_time = int((datetime.now() - timedelta(days=days)).timestamp() * 1000)
print(f"\n📊 Fetching Funding History for {coin}...")
try:
data = self.info.funding_history(coin, start_time, end_time)
if data:
print(f" ✓ Funding Entries: {len(data)}")
if data:
latest = data[-1]
print(f" ✓ Latest Rate: {latest.get('fundingRate', 0)}")
return data
except Exception as e:
print(f" ✗ Error: {e}")
return []
def get_user_funding_history(self, days: int = 7) -> list:
"""
Get user's funding payments history.
Args:
days: Number of days of history (default: 7)
Returns:
List of funding payments
"""
end_time = int(datetime.now().timestamp() * 1000)
start_time = int((datetime.now() - timedelta(days=days)).timestamp() * 1000)
print(f"\n💸 Fetching User Funding Payments (last {days} days)...")
try:
data = self.info.user_funding_history(self.wallet_address, start_time, end_time)
if data:
print(f" ✓ Funding Payments: {len(data)}")
total_funding = sum(float(f.get('usdc', 0)) for f in data)
print(f" ✓ Total Funding P&L: ${total_funding:.2f}")
else:
print(" ⚠ No funding payments found")
return data
except Exception as e:
print(f" ✗ Error: {e}")
return []
def get_user_non_funding_ledger_updates(self, days: int = 7) -> list:
"""
Get non-funding ledger updates (deposits, withdrawals, liquidations).
Args:
days: Number of days of history (default: 7)
Returns:
List of ledger updates
"""
end_time = int(datetime.now().timestamp() * 1000)
start_time = int((datetime.now() - timedelta(days=days)).timestamp() * 1000)
print(f"\n📒 Fetching Ledger Updates (last {days} days)...")
try:
data = self.info.user_non_funding_ledger_updates(self.wallet_address, start_time, end_time)
if data:
print(f" ✓ Ledger Updates: {len(data)}")
# Categorize updates
deposits = [u for u in data if 'deposit' in str(u.get('delta', {})).lower()]
withdrawals = [u for u in data if 'withdraw' in str(u.get('delta', {})).lower()]
print(f" ✓ Deposits: {len(deposits)}, Withdrawals: {len(withdrawals)}")
else:
print(" ⚠ No ledger updates found")
return data
except Exception as e:
print(f" ✗ Error: {e}")
return []
def get_referral_state(self) -> Dict[str, Any]:
"""
Get referral program state for the user.
Returns:
Dict with referral status and earnings
"""
print("\n🎁 Fetching Referral State...")
try:
data = self.info.query_referral_state(self.wallet_address)
if data:
print(f" ✓ Referral Code: {data.get('referralCode', 'N/A')}")
print(f" ✓ Referees: {len(data.get('referees', []))}")
return data
except Exception as e:
print(f" ✗ Error: {e}")
return {}
def get_sub_accounts(self) -> list:
"""
Get list of sub-accounts for the user.
Returns:
List of sub-account addresses
"""
print("\n👥 Fetching Sub-Accounts...")
try:
data = self.info.query_sub_accounts(self.wallet_address)
if data:
print(f" ✓ Sub-Accounts: {len(data)}")
else:
print(" ⚠ No sub-accounts found")
return data
except Exception as e:
print(f" ✗ Error: {e}")
return []
def fetch_all_data(self, save_to_file: bool = True) -> Dict[str, Any]:
"""
Fetch all available data for the wallet.
Args:
save_to_file: If True, save results to JSON file
Returns:
Dict containing all fetched data
"""
print("=" * 80)
print("HYPERLIQUID WALLET DATA FETCHER")
print("=" * 80)
all_data = {
'wallet_address': self.wallet_address,
'timestamp': datetime.now().isoformat(),
'data': {}
}
# Fetch all data sections
all_data['data']['user_state'] = self.get_user_state()
all_data['data']['spot_state'] = self.get_spot_state()
all_data['data']['open_orders'] = self.get_open_orders()
all_data['data']['recent_fills'] = self.get_user_fills(limit=50)
all_data['data']['fills_last_7_days'] = self.get_user_fills_by_time()
all_data['data']['user_fees'] = self.get_user_fees()
all_data['data']['rate_limit'] = self.get_user_rate_limit()
all_data['data']['funding_payments'] = self.get_user_funding_history(days=7)
all_data['data']['ledger_updates'] = self.get_user_non_funding_ledger_updates(days=7)
all_data['data']['referral_state'] = self.get_referral_state()
all_data['data']['sub_accounts'] = self.get_sub_accounts()
# Optional: Fetch funding history for positions
user_state = all_data['data']['user_state']
if user_state and user_state.get('assetPositions'):
all_data['data']['funding_history'] = {}
for position in user_state['assetPositions'][:3]: # First 3 positions
coin = position.get('position', {}).get('coin')
if coin:
all_data['data']['funding_history'][coin] = self.get_funding_history(coin, days=7)
print("\n" + "=" * 80)
print("DATA COLLECTION COMPLETE")
print("=" * 80)
# Save to file
if save_to_file:
filename = f"hyperliquid_wallet_data_{self.wallet_address[:10]}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
with open(filename, 'w') as f:
json.dump(all_data, f, indent=2, default=str)
print(f"\n💾 Data saved to: {filename}")
return all_data
def main():
"""Main execution function."""
if len(sys.argv) < 2:
print("Usage: python hyperliquid_wallet_data.py <wallet_address> [--testnet]")
print("\nExample:")
print(" python hyperliquid_wallet_data.py 0xcd5051944f780a621ee62e39e493c489668acf4d")
sys.exit(1)
wallet_address = sys.argv[1]
use_testnet = '--testnet' in sys.argv
# Validate wallet address format
if not wallet_address.startswith('0x') or len(wallet_address) != 42:
print("❌ Error: Invalid wallet address format")
print(" Address must be in format: 0x followed by 40 hexadecimal characters")
sys.exit(1)
try:
analyzer = HyperliquidWalletAnalyzer(wallet_address, use_testnet=use_testnet)
data = analyzer.fetch_all_data(save_to_file=True)
print("\n✅ All data fetched successfully!")
print(f"\n📊 Summary:")
print(f" - Account Value: ${data['data']['user_state'].get('marginSummary', {}).get('accountValue', 0)}")
print(f" - Open Positions: {len(data['data']['user_state'].get('assetPositions', []))}")
print(f" - Spot Holdings: {len(data['data']['spot_state'].get('balances', []))}")
print(f" - Open Orders: {len(data['data']['open_orders'])}")
print(f" - Recent Fills: {len(data['data']['recent_fills'])}")
except Exception as e:
print(f"\n❌ Fatal Error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
main()

View File

@ -1,367 +0,0 @@
import json
import os
import time
import requests
import logging
import argparse
import sys
from datetime import datetime, timedelta
# --- Configuration ---
# !! IMPORTANT: Update this to your actual Hyperliquid API endpoint !!
API_ENDPOINT = "https://api.hyperliquid.xyz/info"
INPUT_FILE = os.path.join("_data", "wallets_to_track.json")
OUTPUT_FILE = os.path.join("_data", "wallets_info.json")
LOGS_DIR = "_logs"
LOG_FILE = os.path.join(LOGS_DIR, "whale_tracker.log")
# Polling intervals (in seconds)
POLL_INTERVALS = {
'core_data': 10, # 5-15s range
'open_orders': 20, # 15-30s range
'account_metrics': 180, # 1-5m range
'ledger_updates': 600, # 5-15m range
'save_data': 5, # How often to write to wallets_info.json
'reload_wallets': 60 # Check for wallet list changes every 60s
}
class HyperliquidAPI:
"""
Client to handle POST requests to the Hyperliquid info endpoint.
"""
def __init__(self, base_url):
self.base_url = base_url
self.session = requests.Session()
logging.info(f"API Client initialized for endpoint: {base_url}")
def post_request(self, payload):
"""
Internal helper to send POST requests and handle errors.
"""
try:
response = self.session.post(self.base_url, json=payload, timeout=10)
response.raise_for_status() # Raise an exception for bad status codes (4xx or 5xx)
return response.json()
except requests.exceptions.HTTPError as e:
logging.error(f"HTTP Error: {e.response.status_code} for {e.request.url}. Response: {e.response.text}")
except requests.exceptions.ConnectionError as e:
logging.error(f"Connection Error: {e}")
except requests.exceptions.Timeout:
logging.error(f"Request timed out for payload: {payload.get('type')}")
except json.JSONDecodeError:
logging.error(f"Failed to decode JSON response. Response text: {response.text if 'response' in locals() else 'No response text'}")
except Exception as e:
logging.error(f"An unexpected error occurred in post_request: {e}", exc_info=True)
return None
def get_user_state(self, user_address: str):
payload = {"type": "clearinghouseState", "user": user_address}
return self.post_request(payload)
def get_open_orders(self, user_address: str):
payload = {"type": "openOrders", "user": user_address}
return self.post_request(payload)
def get_user_rate_limit(self, user_address: str):
payload = {"type": "userRateLimit", "user": user_address}
return self.post_request(payload)
def get_user_ledger_updates(self, user_address: str, start_time_ms: int, end_time_ms: int):
payload = {
"type": "userNonFundingLedgerUpdates",
"user": user_address,
"startTime": start_time_ms,
"endTime": end_time_ms
}
return self.post_request(payload)
class WalletTracker:
"""
Main class to track wallets, process data, and store results.
"""
def __init__(self, api_client, wallets_to_track):
self.api = api_client
self.wallets = wallets_to_track # This is the list of dicts
self.wallets_by_name = {w['name']: w for w in self.wallets}
self.wallets_data = {
wallet['name']: {"address": wallet['address']} for wallet in self.wallets
}
logging.info(f"WalletTracker initialized for {len(self.wallets)} wallets.")
def reload_wallets(self):
"""
Checks the INPUT_FILE for changes and updates the tracked wallet list.
"""
logging.debug("Reloading wallet list...")
try:
with open(INPUT_FILE, 'r') as f:
new_wallets_list = json.load(f)
if not isinstance(new_wallets_list, list):
logging.warning(f"Failed to reload '{INPUT_FILE}': content is not a list.")
return
new_wallets_by_name = {w['name']: w for w in new_wallets_list}
old_names = set(self.wallets_by_name.keys())
new_names = set(new_wallets_by_name.keys())
added_names = new_names - old_names
removed_names = old_names - new_names
if not added_names and not removed_names:
logging.debug("Wallet list is unchanged.")
return # No changes
# Update internal wallet list
self.wallets = new_wallets_list
self.wallets_by_name = new_wallets_by_name
# Add new wallets to wallets_data
for name in added_names:
self.wallets_data[name] = {"address": self.wallets_by_name[name]['address']}
logging.info(f"Added new wallet to track: {name}")
# Remove old wallets from wallets_data
for name in removed_names:
if name in self.wallets_data:
del self.wallets_data[name]
logging.info(f"Removed wallet from tracking: {name}")
logging.info(f"Wallet list reloaded. Tracking {len(self.wallets)} wallets.")
except (FileNotFoundError, json.JSONDecodeError, ValueError) as e:
logging.error(f"Failed to reload and parse '{INPUT_FILE}': {e}")
except Exception as e:
logging.error(f"Unexpected error during wallet reload: {e}", exc_info=True)
def calculate_core_metrics(self, state_data: dict) -> dict:
"""
Performs calculations based on user_state data.
"""
if not state_data or 'crossMarginSummary' not in state_data:
logging.warning("Core state data is missing 'crossMarginSummary'.")
return {"raw_state": state_data}
summary = state_data['crossMarginSummary']
account_value = float(summary.get('accountValue', 0))
margin_used = float(summary.get('totalMarginUsed', 0))
# Calculations
margin_utilization = (margin_used / account_value) if account_value > 0 else 0
available_margin = account_value - margin_used
total_position_value = 0
if 'assetPositions' in state_data:
for pos in state_data.get('assetPositions', []):
try:
# Use 'value' for position value
pos_value_str = pos.get('position', {}).get('value', '0')
total_position_value += float(pos_value_str)
except (ValueError, TypeError):
logging.warning(f"Could not parse position value: {pos.get('position', {}).get('value')}")
continue
portfolio_leverage = (total_position_value / account_value) if account_value > 0 else 0
# Return calculated metrics alongside raw data
return {
"raw_state": state_data,
"account_value": account_value,
"margin_used": margin_used,
"margin_utilization": margin_utilization,
"available_margin": available_margin,
"total_position_value": total_position_value,
"portfolio_leverage": portfolio_leverage
}
def poll_core_data(self):
logging.debug("Polling Core Data...")
# Use self.wallets which is updated by reload_wallets
for wallet in self.wallets:
name = wallet['name']
address = wallet['address']
state_data = self.api.get_user_state(address)
if state_data:
calculated_data = self.calculate_core_metrics(state_data)
# Ensure wallet hasn't been removed by a concurrent reload
if name in self.wallets_data:
self.wallets_data[name]['core_state'] = calculated_data
time.sleep(0.1) # Avoid bursting requests
def poll_open_orders(self):
logging.debug("Polling Open Orders...")
for wallet in self.wallets:
name = wallet['name']
address = wallet['address']
orders_data = self.api.get_open_orders(address)
if orders_data:
# TODO: Add calculations for 'pending_margin_required' if logic is available
if name in self.wallets_data:
self.wallets_data[name]['open_orders'] = {"raw_orders": orders_data}
time.sleep(0.1)
def poll_account_metrics(self):
logging.debug("Polling Account Metrics...")
for wallet in self.wallets:
name = wallet['name']
address = wallet['address']
metrics_data = self.api.get_user_rate_limit(address)
if metrics_data:
if name in self.wallets_data:
self.wallets_data[name]['account_metrics'] = metrics_data
time.sleep(0.1)
def poll_ledger_updates(self):
logging.debug("Polling Ledger Updates...")
end_time_ms = int(datetime.now().timestamp() * 1000)
start_time_ms = int((datetime.now() - timedelta(minutes=15)).timestamp() * 1000)
for wallet in self.wallets:
name = wallet['name']
address = wallet['address']
ledger_data = self.api.get_user_ledger_updates(address, start_time_ms, end_time_ms)
if ledger_data:
if name in self.wallets_data:
self.wallets_data[name]['ledger_updates'] = ledger_data
time.sleep(0.1)
def save_data_to_json(self):
"""
Atomically writes the current wallet data to the output JSON file.
(No longer needs cleaning logic)
"""
logging.debug(f"Saving data to {OUTPUT_FILE}...")
temp_file = OUTPUT_FILE + ".tmp"
try:
# Save the data
with open(temp_file, 'w', encoding='utf-8') as f:
# self.wallets_data is automatically kept clean by reload_wallets
json.dump(self.wallets_data, f, indent=2)
# Atomic rename (move)
os.replace(temp_file, OUTPUT_FILE)
except (IOError, json.JSONDecodeError) as e:
logging.error(f"Failed to write wallet data to file: {e}")
except Exception as e:
logging.error(f"An unexpected error occurred during file save: {e}")
if os.path.exists(temp_file):
os.remove(temp_file)
class WhaleTrackerRunner:
"""
Manages the polling loop using last-run timestamps instead of a complex scheduler.
"""
def __init__(self, api_client, wallets, shared_whale_data_dict=None): # Kept arg for compatibility
self.tracker = WalletTracker(api_client, wallets)
self.last_poll_times = {key: 0 for key in POLL_INTERVALS}
self.poll_intervals = POLL_INTERVALS
logging.info("WhaleTrackerRunner initialized to save to JSON file.")
def update_shared_data(self):
"""
This function is no longer called by the run loop.
It's kept here to prevent errors if imported elsewhere, but is now unused.
"""
logging.debug("No shared dict, saving data to JSON file.")
self.tracker.save_data_to_json()
def run(self):
logging.info("Starting main polling loop...")
while True:
try:
now = time.time()
if now - self.last_poll_times['reload_wallets'] > self.poll_intervals['reload_wallets']:
self.tracker.reload_wallets()
self.last_poll_times['reload_wallets'] = now
if now - self.last_poll_times['core_data'] > self.poll_intervals['core_data']:
self.tracker.poll_core_data()
self.last_poll_times['core_data'] = now
if now - self.last_poll_times['open_orders'] > self.poll_intervals['open_orders']:
self.tracker.poll_open_orders()
self.last_poll_times['open_orders'] = now
if now - self.last_poll_times['account_metrics'] > self.poll_intervals['account_metrics']:
self.tracker.poll_account_metrics()
self.last_poll_times['account_metrics'] = now
if now - self.last_poll_times['ledger_updates'] > self.poll_intervals['ledger_updates']:
self.tracker.poll_ledger_updates()
self.last_poll_times['ledger_updates'] = now
if now - self.last_poll_times['save_data'] > self.poll_intervals['save_data']:
self.tracker.save_data_to_json() # <-- NEW
self.last_poll_times['save_data'] = now
# Sleep for a short duration to prevent busy-waiting
time.sleep(1)
except Exception as e:
logging.critical(f"Unhandled exception in main loop: {e}", exc_info=True)
time.sleep(10)
def setup_logging(log_level_str: str, process_name: str):
"""Configures logging for the script."""
if not os.path.exists(LOGS_DIR):
try:
os.makedirs(LOGS_DIR)
except OSError as e:
print(f"Failed to create logs directory {LOGS_DIR}: {e}")
return
level_map = {
'debug': logging.DEBUG,
'normal': logging.INFO,
'off': logging.NOTSET
}
log_level = level_map.get(log_level_str.lower(), logging.INFO)
if log_level == logging.NOTSET:
return
handlers_list = [logging.FileHandler(LOG_FILE, mode='a')]
if sys.stdout.isatty():
handlers_list.append(logging.StreamHandler(sys.stdout))
logging.basicConfig(
level=log_level,
format=f"%(asctime)s.%(msecs)03d | {process_name:<20} | %(levelname)-8s | %(message)s",
datefmt='%Y-%m-%d %H:%M:%S',
handlers=handlers_list
)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Hyperliquid Whale Tracker")
parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug'])
args = parser.parse_args()
setup_logging(args.log_level, "WhaleTracker")
# Load wallets to track
wallets_to_track = []
try:
with open(INPUT_FILE, 'r') as f:
wallets_to_track = json.load(f)
if not isinstance(wallets_to_track, list) or not wallets_to_track:
raise ValueError(f"'{INPUT_FILE}' is empty or not a list.")
except (FileNotFoundError, json.JSONDecodeError, ValueError) as e:
logging.critical(f"Failed to load '{INPUT_FILE}': {e}. Exiting.")
sys.exit(1)
# Initialize API client
api_client = HyperliquidAPI(base_url=API_ENDPOINT)
# Initialize and run the tracker
runner = WhaleTrackerRunner(api_client, wallets_to_track, shared_whale_data_dict=None)
try:
runner.run()
except KeyboardInterrupt:
logging.info("Whale Tracker shutting down.")
sys.exit(0)