diff --git a/.dockerignore b/.dockerignore index 2605f78..3335263 100644 --- a/.dockerignore +++ b/.dockerignore @@ -12,3 +12,6 @@ agents/ secrets/ .env.docker .env +clp_hedger.log +clp_hedger/hedge_status.json +backups/ diff --git a/.gitignore b/.gitignore index 0c358e4..f352c68 100644 --- a/.gitignore +++ b/.gitignore @@ -22,6 +22,9 @@ _data/*.json # Ignore all log files _logs/ +# Ignore backups +backups/ + # --- SDK --- # Ignore all contents of the sdk directory sdk/ @@ -30,6 +33,15 @@ sdk/ # Ignore custom agents directory 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 .temp/ diff --git a/DOCKER_MIGRATION_GUIDE.md b/DOCKER_MIGRATION_GUIDE.md new file mode 100644 index 0000000..85133c6 --- /dev/null +++ b/DOCKER_MIGRATION_GUIDE.md @@ -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 diff --git a/GEMINI.md b/GEMINI.md index 6457ac0..b05583b 100644 --- a/GEMINI.md +++ b/GEMINI.md @@ -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. * **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). \ No newline at end of file diff --git a/MIGRATION_PLAN.md b/MIGRATION_PLAN.md index 78169a8..d047e24 100644 --- a/MIGRATION_PLAN.md +++ b/MIGRATION_PLAN.md @@ -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 ``` @@ -79,9 +79,9 @@ wal_buffers = 4MB 2. `resampler.py` — `sqlite3` → `db.py` 3. `data_fetcher.py` — `sqlite3` → `db.py` 4. `fetch_history.py` — `sqlite3` → `db.py` -5. `import_csv.py` — `sqlite3` → `db.py` +5. `scripts/import_csv.py` — `sqlite3` → `db.py` 6. `indicators.py` — `sqlite3` → `psycopg2` -7. `base_strategy.py` — `sqlite3` → `psycopg2` +7. `strategies/base_strategy.py` — `sqlite3` → `psycopg2` ## TODO List @@ -90,34 +90,34 @@ wal_buffers = 4MB - [x] Add `psycopg2-binary` to `requirements.txt` ### Phase 2: Modify Data Collection Components -- [ ] Modify `live_candle_fetcher.py` — replace `sqlite3.connect()` with `db.get_connection()`, `INSERT OR REPLACE` with `db.upsert_candles()`, sanitize table names -- [ ] Modify `resampler.py` — replace `sqlite3` with `db.py`, `INSERT OR REPLACE` with `db.upsert_candles()`, `?` → `%s` -- [ ] Modify `data_fetcher.py` — replace `sqlite3` with `db.py`, `to_sql()` → `db.upsert_candles()` -- [ ] Modify `fetch_history.py` — replace `sqlite3` with `db.py` -- [ ] Modify `import_csv.py` — replace `sqlite3` with `db.py`, `to_sql()` → `db.upsert_candles()` +- [x] Modify `live_candle_fetcher.py` — replace `sqlite3.connect()` with `db.get_connection()`, `INSERT OR REPLACE` with `db.upsert_candles()`, sanitize table names +- [x] Modify `resampler.py` — replace `sqlite3` with `db.py`, `INSERT OR REPLACE` with `db.upsert_candles()`, `?` → `%s` +- [x] Modify `data_fetcher.py` — replace `sqlite3` with `db.py`, `to_sql()` → `db.upsert_candles()` +- [x] Modify `fetch_history.py` — replace `sqlite3` with `db.py` +- [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 -- [ ] Create `scripts/resampler_loop.py` — wraps resampler in a while loop with 60s sleep -- [ ] Create `scripts/gap_detector.py` — detects gaps in 1m data, backfills via HTTP API -- [ ] Create `scripts/backup_runner.py` — daily pg_dump with 7-day retention -- [ ] Create `scripts/cron_scheduler.py` — schedules data_fetcher, fetch_history, gap_detector, backup +- [x] Create `scripts/resampler_loop.py` — wraps resampler in a while loop with 60s sleep +- [x] Create `scripts/gap_detector.py` — detects gaps in 1m data, backfills via HTTP API +- [x] Create `scripts/backup_runner.py` — daily pg_dump with 7-day retention +- [x] Create `scripts/cron_scheduler.py` — schedules data_fetcher, fetch_history, gap_detector, backup ### Phase 4: Docker Setup -- [ ] Create `Dockerfile` (python:3.11-slim + supervisor + psycopg2-binary) -- [ ] Create `docker-compose.yml` (postgres + data-collector services) -- [ ] Create `supervisord.conf` (live_candle_fetcher, resampler_loop, indicators_fetcher, cron_scheduler) -- [ ] Create `postgres/postgresql.conf` (tuned for 4GB RAM) -- [ ] Create `.dockerignore` -- [ ] Create `.env.docker.example` -- [ ] Create `secrets/pg_password.txt.example` -- [ ] Update `.gitignore` +- [x] Create `Dockerfile` (python:3.11-slim + supervisor + psycopg2-binary) +- [x] Create `docker-compose.yml` (postgres + data-collector services) +- [x] Create `supervisord.conf` (live_candle_fetcher, resampler_loop, indicators_fetcher, cron_scheduler) +- [x] Create `postgres/postgresql.conf` (tuned for 4GB RAM) +- [x] Create `.dockerignore` +- [x] Create `.env.docker.example` +- [x] Create `secrets/pg_password.txt.example` +- [x] Update `.gitignore` ### Phase 5: Host-Side Updates - [ ] Modify `indicators.py` on host — connect to `localhost:5432` -- [ ] Modify `base_strategy.py` on host — connect to `localhost:5432` +- [ ] Modify `strategies/base_strategy.py` on host — connect to `localhost:5432` ### Phase 6: Migration Tool -- [ ] Create `migrate_sqlite_to_pg.py` — reads from SQLite, writes to PostgreSQL +- [x] Create `migrate_sqlite_to_pg.py` — reads from SQLite, writes to PostgreSQL ### Phase 7: Testing & Deployment - [ ] Commit and push to remote diff --git a/_data/backtesting_conf.json.example b/_data/backtesting_conf.json.example new file mode 100644 index 0000000..aec37b7 --- /dev/null +++ b/_data/backtesting_conf.json.example @@ -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 + } + } + } +} diff --git a/_data/coin_precision.json b/_data/coin_precision.json index db6470c..416f67d 100644 --- a/_data/coin_precision.json +++ b/_data/coin_precision.json @@ -4,6 +4,7 @@ "AAVE": 2, "ACE": 2, "ADA": 0, + "AERO": 0, "AI": 1, "AI16Z": 1, "AIXBT": 0, @@ -20,11 +21,11 @@ "ATOM": 2, "AVAX": 2, "AVNT": 0, + "AXS": 1, + "AZTEC": 0, "BABY": 0, "BADGER": 1, "BANANA": 1, - "BASH": 0, - "BATH": 0, "BCH": 3, "BERA": 1, "BIGTIME": 0, @@ -40,13 +41,17 @@ "BTC": 5, "CAKE": 1, "CANTO": 0, + "CASHCAT": 0, "CATI": 0, + "CC": 0, "CELO": 0, "CFX": 0, "CHILLGUY": 0, + "CHIP": 0, "COMP": 2, "CRV": 1, "CYBER": 1, + "DASH": 2, "DOGE": 0, "DOOD": 0, "DOT": 1, @@ -61,6 +66,7 @@ "FARTCOIN": 1, "FET": 0, "FIL": 1, + "FOGO": 0, "FRIEND": 1, "FTM": 0, "FTT": 1, @@ -70,6 +76,7 @@ "GMT": 0, "GMX": 2, "GOAT": 0, + "GRAM": 0, "GRASS": 1, "GRIFFAIN": 0, "HBAR": 0, @@ -78,6 +85,7 @@ "HPOS": 0, "HYPE": 2, "HYPER": 0, + "ICP": 1, "ILV": 2, "IMX": 1, "INIT": 0, @@ -96,6 +104,7 @@ "LINEA": 0, "LINK": 1, "LISTA": 0, + "LIT": 0, "LOOM": 0, "LTC": 2, "MANTA": 1, @@ -163,11 +172,13 @@ "SCR": 1, "SEI": 0, "SHIA": 0, + "SKR": 0, "SKY": 0, "SNX": 1, "SOL": 2, "SOPH": 0, "SPX": 1, + "STABLE": 0, "STBL": 0, "STG": 0, "STRAX": 0, @@ -201,10 +212,9 @@ "WLFI": 0, "XAI": 1, "XLM": 0, + "XMR": 3, "XPL": 0, "XRP": 0, - "XYZ:CLUSD": 2, - "xyz:BRENTOIL": 2, "YGG": 0, "YZY": 0, "ZEC": 2, @@ -220,6 +230,5 @@ "kLUNC": 0, "kNEIRO": 1, "kPEPE": 0, - "kSHIB": 0, - "xyz:CLUSD": 2 + "kSHIB": 0 } \ No newline at end of file diff --git a/_data/coin_precision.json.example b/_data/coin_precision.json.example new file mode 100644 index 0000000..8b2821c --- /dev/null +++ b/_data/coin_precision.json.example @@ -0,0 +1,10 @@ +{ + "BTC": 5, + "ETH": 4, + "SOL": 2, + "BNB": 3, + "HYPE": 2, + "SUI": 1, + "0G": 0, + "2Z": 0 +} diff --git a/_data/strategies.json.example b/_data/strategies.json.example new file mode 100644 index 0000000..e02a5a3 --- /dev/null +++ b/_data/strategies.json.example @@ -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 + } + } + } + } +} diff --git a/address_monitor.py b/address_monitor.py deleted file mode 100644 index ced5e44..0000000 --- a/address_monitor.py +++ /dev/null @@ -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() - diff --git a/backtester.py b/backtester.py deleted file mode 100644 index d0c597f..0000000 --- a/backtester.py +++ /dev/null @@ -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.") - diff --git a/base_strategy.py b/base_strategy.py deleted file mode 100644 index 7c3d359..0000000 --- a/base_strategy.py +++ /dev/null @@ -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 diff --git a/basic_ws.py b/basic_ws.py deleted file mode 100644 index bdd375d..0000000 --- a/basic_ws.py +++ /dev/null @@ -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() \ No newline at end of file diff --git a/clp_hedger.log b/clp_hedger.log deleted file mode 100644 index 7767b59..0000000 --- a/clp_hedger.log +++ /dev/null @@ -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. diff --git a/clp_hedger/hedge_status.json b/clp_hedger/hedge_status.json deleted file mode 100644 index 2098378..0000000 --- a/clp_hedger/hedge_status.json +++ /dev/null @@ -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 - } -] \ No newline at end of file diff --git a/data_fetcher_old.py b/data_fetcher_old.py deleted file mode 100644 index 90b3e73..0000000 --- a/data_fetcher_old.py +++ /dev/null @@ -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() - diff --git a/db.py b/db.py index a3b6ac8..415481d 100644 --- a/db.py +++ b/db.py @@ -74,6 +74,8 @@ def upsert_candles(conn, table_name, records): if not records: return 0 + records = list({r[1]: r for r in records}.values()) + with conn.cursor() as cur: execute_values( cur, diff --git a/docker-compose.yml b/docker-compose.yml index a82678b..64c64dc 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,5 +1,3 @@ -version: "3.8" - services: postgres: image: postgres:15-alpine @@ -8,29 +6,37 @@ services: environment: POSTGRES_DB: hyper POSTGRES_USER: hyper - POSTGRES_PASSWORD: ${POSTGRES_PASSWORD} + POSTGRES_PASSWORD: kaqpaaoi0 volumes: - pg_data:/var/lib/postgresql/data - ./postgres/postgresql.conf:/etc/postgresql/postgresql.conf command: postgres -c config_file=/etc/postgresql/postgresql.conf ports: - - "5432:5432" + - "5433:5432" networks: - hyper_net + healthcheck: + test: ["CMD-SHELL", "pg_isready -U [secret] -d [secret]"] + interval: 10s + timeout: 5s + retries: 5 data-collector: - build: . + image: hyper-data-collector:latest container_name: hyper_data restart: unless-stopped depends_on: - - postgres + postgres: + condition: service_healthy env_file: - .env.docker + environment: + - PYTHONPATH=/app volumes: - ./_data:/app/_data - ./_logs:/app/_logs - ./secrets:/app/secrets - - /volume1/docker/hyper/backups:/backups + - /volume2/docker/hyper/backups:/backups networks: - hyper_net @@ -40,3 +46,6 @@ volumes: networks: hyper_net: driver: bridge + ipam: + config: + - subnet: 172.22.0.0/16 \ No newline at end of file diff --git a/fetch_history.py b/fetch_history.py index 1a3d307..82f9a48 100644 --- a/fetch_history.py +++ b/fetch_history.py @@ -1,3 +1,4 @@ +import os import requests import json import db diff --git a/indicators.py b/indicators.py index b9b0b99..339680f 100644 --- a/indicators.py +++ b/indicators.py @@ -1,7 +1,7 @@ """ Indicator calculation module. 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. """ @@ -17,7 +17,7 @@ import numpy as np 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. """ diff --git a/indicators_fetcher.py b/indicators_fetcher.py index 4cdf200..87ebaa0 100644 --- a/indicators_fetcher.py +++ b/indicators_fetcher.py @@ -2,7 +2,7 @@ Indicators Data Fetcher A standalone process that runs in a loop to compute financial indicators -(ratios, prices, MAs, RSI, custom) from SQLite candle data and save + (ratios, prices, MAs, RSI, custom) from PostgreSQL candle data and save the results to a JSON status file for the main dashboard to display. Follows the same pattern as dashboard_data_fetcher.py. diff --git a/list_latest_candles.py b/list_latest_candles.py new file mode 100644 index 0000000..31a7d86 --- /dev/null +++ b/list_latest_candles.py @@ -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) diff --git a/market_old.py b/market_old.py deleted file mode 100644 index ea81112..0000000 --- a/market_old.py +++ /dev/null @@ -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 - diff --git a/position_monitor.py b/position_monitor.py deleted file mode 100644 index b017dce..0000000 --- a/position_monitor.py +++ /dev/null @@ -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() diff --git a/requirements.txt b/requirements.txt index 816426a..de2cde5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -29,7 +29,6 @@ msgpack==1.1.2 multidict==6.7.0 numpy==2.0.2 pandas==2.3.3 -parsimonious==0.10.0 propcache==0.4.1 pycares==4.11.0 pycparser==2.23 @@ -51,6 +50,6 @@ typing_extensions==4.15.0 tzdata==2025.2 urllib3==1.26.20 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 psycopg2-binary==2.9.9 diff --git a/resampler.py b/resampler.py index 098f694..f807211 100644 --- a/resampler.py +++ b/resampler.py @@ -2,17 +2,20 @@ import argparse import logging import os import sys +import warnings import db import pandas as pd import json from datetime import datetime, timezone, timedelta +warnings.filterwarnings("ignore", message="pandas only supports SQLAlchemy") + # Assuming logging_utils.py is in the same directory from logging_utils import setup_logging 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, preventing data duplication. """ @@ -79,13 +82,13 @@ class Resampler: logging.debug(f"Processing {len(self.coins_to_process)} coins...") for coin in self.coins_to_process: - logging.debug(f"--- Processing {coin} ---") + logging.info(f"--- Processing {coin} ---") try: for tf_name, tf_code in self.timeframes.items(): target_table_name = db.sanitize_table_name(coin, tf_name) 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) @@ -119,8 +122,8 @@ class Resampler: records_to_upsert.append(( index.strftime('%Y-%m-%d %H:%M:%S'), int(index.timestamp() * 1000), # Generate timestamp_ms - row['open'], row['high'], row['low'], row['close'], - row['volume'], row['number_of_trades'] + float(row['open']), float(row['high']), float(row['low']), float(row['close']), + float(row['volume']), int(row['number_of_trades']) )) db.upsert_candles(conn, target_table_name, records_to_upsert) @@ -225,7 +228,7 @@ def parse_timeframes(tf_strings: list) -> dict: 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("--timeframes", nargs='+', required=True, help="List of timeframes to generate.") parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug']) diff --git a/review.md b/review.md index d0fe34c..f37e17c 100644 --- a/review.md +++ b/review.md @@ -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 @@ -29,9 +78,8 @@ The following sections detail recommendations for improving configuration manage ### 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:** - - 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. @@ -41,39 +89,3 @@ The following sections detail recommendations for improving configuration manage - **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). diff --git a/check_wtioil.py b/scripts/check_wtioil.py similarity index 100% rename from check_wtioil.py rename to scripts/check_wtioil.py diff --git a/create_agent.py b/scripts/create_agent.py similarity index 100% rename from create_agent.py rename to scripts/create_agent.py diff --git a/del_market_cap_tables.py b/scripts/del_market_cap_tables.py similarity index 100% rename from del_market_cap_tables.py rename to scripts/del_market_cap_tables.py diff --git a/fix_timestamps.py b/scripts/fix_timestamps.py similarity index 100% rename from fix_timestamps.py rename to scripts/fix_timestamps.py diff --git a/import_csv.py b/scripts/import_csv.py similarity index 99% rename from import_csv.py rename to scripts/import_csv.py index 7298969..7f6b362 100644 --- a/import_csv.py +++ b/scripts/import_csv.py @@ -11,7 +11,7 @@ from logging_utils import setup_logging 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. """ @@ -139,7 +139,7 @@ class CsvImporter: 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("--coin", default="BTC", help="The coin symbol for this data (e.g., BTC).") parser.add_argument( diff --git a/list_coins.py b/scripts/list_coins.py similarity index 100% rename from list_coins.py rename to scripts/list_coins.py diff --git a/!migrate_to_sqlite.py b/scripts/migrate_to_sqlite.py similarity index 100% rename from !migrate_to_sqlite.py rename to scripts/migrate_to_sqlite.py diff --git a/strategy_sma_cross.py b/strategy_sma_cross.py deleted file mode 100644 index e47331e..0000000 --- a/strategy_sma_cross.py +++ /dev/null @@ -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) - diff --git a/strategy_template.py b/strategy_template.py deleted file mode 100644 index ba4b9ed..0000000 --- a/strategy_template.py +++ /dev/null @@ -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) - diff --git a/trade_log.py b/trade_log.py deleted file mode 100644 index 6360dd5..0000000 --- a/trade_log.py +++ /dev/null @@ -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}") diff --git a/wallet_data.py b/wallet_data.py deleted file mode 100644 index 0188724..0000000 --- a/wallet_data.py +++ /dev/null @@ -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 - -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 [--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() \ No newline at end of file diff --git a/whale_tracker.py b/whale_tracker.py deleted file mode 100644 index 69fc29a..0000000 --- a/whale_tracker.py +++ /dev/null @@ -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) -