ditus/uniswap_auto_clp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: Standardize CLP Manager and Hedger modules & cleanup

Browse Source 
- **clp_manager.py**: Renamed from 'uniswap_manager.py'. Standardized logic for Uniswap V3 liquidity provision.
- **clp_hedger.py**: Renamed from 'unified_hedger.py'. Consolidated hedging logic including Delta Calculation fixes, EAC (Edge Avoidance), and Fishing order implementation.
- **Cleanup**: Removed legacy 'aerodrome' folder and tools.
- **Monitoring**: Added Telegram monitoring scripts.
- **Config**: Updated gitignore to exclude market data CSVs.
This commit is contained in:
DiTus
2025-12-31 11:09:33 +01:00
parent 69fbf389c8
commit b22fdcf741
31 changed files with 3499 additions and 6869 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
tools/.env.example Normal file
View File

@ -0,0 +1,18 @@
# Base Chain Configuration
BASE_RPC_URL=https://mainnet.base.org # or your preferred Base RPC provider
# Wallet Configuration
MAIN_WALLET_PRIVATE_KEY=your_private_key_here
# or
PRIVATE_KEY=your_private_key_here
# Telegram Notifications (Optional)
TELEGRAM_MONITOR_ENABLED=False
TELEGRAM_BOT_TOKEN=your_telegram_bot_token
TELEGRAM_CHAT_ID=your_telegram_chat_id
TELEGRAM_CHECK_INTERVAL_SECONDS=60
TELEGRAM_TIMEOUT_SECONDS=10
TELEGRAM_STATE_FILE=telegram_monitor_state.json
# Optional: Custom hedge status file path
HEDGE_STATUS_FILE=hedge_status.json

194
tools/README_TELEGRAM.md Normal file
View File

@ -0,0 +1,194 @@
# CLP Telegram Notification Monitor
## Overview
Standalone Python script that monitors your CLP (Concentrated Liquidity Pool) positions and sends Telegram notifications when new positions are opened.
## Features
- 🔍 **File-based Monitoring**: Watches `hedge_status.json` every 60 seconds
- 📊 **Rich Notifications**: Shows last closed position vs currently opened position
- 🔧 **Zero Integration**: No modifications to existing trading logic required
- 🛡️ **Safe Operation**: Graceful error handling, won't affect trading system
- 📈 **Performance Tracking**: Compares entry prices, values, and hedge PnL
## Setup Instructions
### 1. Install Dependencies
```bash
pip install requests python-dotenv
```
### 2. Create Telegram Bot
1. Open Telegram and search for **@BotFather**
2. Send: `/newbot`
3. Choose a name for your bot (e.g., "CLP Position Monitor")
4. Choose a username (must end with 'bot', e.g., "clp_monitor_bot")
5. Copy the **HTTP API token** from BotFather
### 3. Get Your Chat ID
1. Send any message to your new bot first
2. Run the setup script: `python tools/telegram_setup.py`
3. Choose option "2. Get Chat ID" and enter your bot token
4. Copy your Chat ID
### 4. Configure Environment
Add to your `.env` file:
```bash
# Enable Telegram notifications
TELEGRAM_MONITOR_ENABLED=True
# Your bot credentials
TELEGRAM_BOT_TOKEN=123456789:ABCdefGHIjklMNOpqrsTUVwxyz
TELEGRAM_CHAT_ID=123456789
# Optional settings
TELEGRAM_CHECK_INTERVAL_SECONDS=60
TELEGRAM_TIMEOUT_SECONDS=10
TELEGRAM_STATE_FILE=telegram_monitor_state.json
HEDGE_STATUS_FILE=hedge_status.json
```
### 5. Run the Monitor
```bash
python tools/telegram_monitor.py
```
## Usage
### Running as Background Process
**Option 1: Simple Background**
```bash
nohup python tools/telegram_monitor.py > logs/telegram_monitor.log 2>&1 &
```
**Option 2: Screen Session**
```bash
screen -S telegram_monitor
python tools/telegram_monitor.py
# Press Ctrl+A, D to detach
```
**Option 3: Systemd Service** (Linux)
Create `/etc/systemd/system/clp-telegram-monitor.service`:
```ini
[Unit]
Description=CLP Telegram Monitor
After=network.target
[Service]
Type=simple
User=your_username
WorkingDirectory=/path/to/uniswap_auto_clp
ExecStart=/usr/bin/python3 tools/telegram_monitor.py
Restart=always
RestartSec=30
[Install]
WantedBy=multi-user.target
```
Enable and start:
```bash
sudo systemctl enable clp-telegram-monitor
sudo systemctl start clp-telegram-monitor
```
## Message Format Example
When a new position opens, you'll receive:
```
🚀 NEW CLP POSITION DETECTED
📊 LAST CLOSED POSITION:
• Token ID: 5171687
• Entry: $3043.86
• Target Value: $1985.02
• Duration: 2h 14m
• Hedge PnL: -$20.99
• Hedge Fees: $2.30
🔥 CURRENTLY OPENED:
• Token ID: 5173016
• Entry: $2993.72
• Target Value: $1935.04
• Range: $2849.80 - $3140.07
• Initial: 0.3181 ETH + 982.71 USDC
• Time: 2h 15m ago
📈 PERFORMANCE COMPARISON:
• Entry Change: -$50.14 (-1.65%)
• Value Change: -$49.98 (-2.52%)
• Hedge PnL Trend: -$20.99 → $2.75 (+$23.74)
```
## Configuration Options
| Variable | Default | Description |
|----------|----------|-------------|
| `TELEGRAM_MONITOR_ENABLED` | `False` | Enable/disable notifications |
| `TELEGRAM_BOT_TOKEN` | Required | Your bot's HTTP API token |
| `TELEGRAM_CHAT_ID` | Required | Your personal chat ID |
| `TELEGRAM_CHECK_INTERVAL_SECONDS` | `60` | How often to check for changes |
| `TELEGRAM_TIMEOUT_SECONDS` | `10` | Telegram API timeout |
| `TELEGRAM_STATE_FILE` | `telegram_monitor_state.json` | Internal state tracking |
| `HEDGE_STATUS_FILE` | `hedge_status.json` | File to monitor |
## Troubleshooting
### Bot Not Receiving Messages
- Ensure you've sent a message to your bot first
- Check that `TELEGRAM_CHAT_ID` is correct (no quotes)
- Verify bot token has no extra spaces
### No Notifications When Position Opens
- Check `TELEGRAM_MONITOR_ENABLED=True`
- Verify the script is running: `ps aux | grep telegram_monitor`
- Check logs: `tail -f logs/telegram_monitor.log`
- Ensure `hedge_status.json` path is correct
### Telegram API Errors
- Verify bot token format (should have colon: `123:ABC`)
- Check internet connectivity
- Try running setup script to test connection
### File Not Found Errors
- Make sure you're running from project root directory
- Verify `hedge_status.json` exists and has correct permissions
- Check that your trading system is generating the status file
## Security Notes
- **Never share your bot token** - it's like a password
- **Store credentials in `.env`** - never commit to git
- **Limit bot permissions** - only send messages, no admin rights
- **Monitor logs regularly** for any unusual activity
## Files Created
- `tools/telegram_monitor.py` - Main monitoring script
- `tools/telegram_setup.py` - Setup helper script
- `tools/.env.example` - Environment template
- `telegram_monitor_state.json` - Internal state (auto-created)
- `logs/telegram_monitor.log` - Monitor logs (auto-created)
## Integration Notes
This monitor is **completely standalone** and:
- ✅ Does not modify your trading logic
- ✅ Does not interfere with uniswap_manager.py
- ✅ Does not affect clp_hedger.py
- ✅ Safe to run alongside existing processes
- ✅ Can be stopped/started independently
- ✅ Uses minimal system resources
The monitor simply reads the status file that your existing trading system already generates, making it safe and non-intrusive.

394
tools/telegram_monitor.py Normal file
View File

@ -0,0 +1,394 @@
#!/usr/bin/env python3
"""
Telegram Monitor for CLP Position Notifications
Monitors hedge_status.json file for new position openings and sends Telegram notifications
"""
import os
import sys
import time
import json
import logging
import hashlib
import requests
from datetime import datetime
from decimal import Decimal
from typing import Optional, Dict, List, Tuple, Any
from dotenv import load_dotenv
# --- SETUP PROJECT PATH ---
current_dir = os.path.dirname(os.path.abspath(__file__))
project_root = os.path.dirname(current_dir)
sys.path.append(current_dir)
# --- LOGGING SETUP ---
os.makedirs(os.path.join(current_dir, 'logs'), exist_ok=True)
class UnixMsLogFilter(logging.Filter):
def filter(self, record):
record.unix_ms = int(record.created * 1000)
return True
logger = logging.getLogger("TELEGRAM_MONITOR")
logger.setLevel(logging.INFO)
logger.propagate = False
logger.handlers.clear()
# Console handler
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
console_handler.setFormatter(console_formatter)
logger.addHandler(console_handler)
# File handler
file_handler = logging.FileHandler(os.path.join(current_dir, 'logs', 'telegram_monitor.log'), encoding='utf-8')
file_handler.setLevel(logging.INFO)
file_handler.addFilter(UnixMsLogFilter())
file_formatter = logging.Formatter('%(unix_ms)d, %(asctime)s - %(name)s - %(levelname)s - %(message)s')
file_handler.setFormatter(file_formatter)
logger.addHandler(file_handler)
# --- CONFIGURATION ---
load_dotenv(os.path.join(current_dir, '.env'))
TELEGRAM_ENABLED = os.getenv('TELEGRAM_MONITOR_ENABLED', 'False').lower() == 'true'
TELEGRAM_BOT_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN', '')
TELEGRAM_CHAT_ID = os.getenv('TELEGRAM_CHAT_ID', '')
TELEGRAM_CHECK_INTERVAL = int(os.getenv('TELEGRAM_CHECK_INTERVAL_SECONDS', '60'))
TELEGRAM_STATE_FILE = os.getenv('TELEGRAM_STATE_FILE', 'telegram_monitor_state.json')
TELEGRAM_TIMEOUT = int(os.getenv('TELEGRAM_TIMEOUT_SECONDS', '10'))
HEDGE_STATUS_FILE = os.getenv('HEDGE_STATUS_FILE', 'hedge_status.json')
class TelegramNotifier:
"""Handles Telegram API communication"""
def __init__(self, bot_token: str, chat_id: str):
self.bot_token = bot_token
self.chat_id = chat_id
self.base_url = f"https://api.telegram.org/bot{bot_token}"
def test_connection(self) -> bool:
"""Test Telegram bot connection"""
try:
url = f"{self.base_url}/getMe"
response = requests.get(url, timeout=TELEGRAM_TIMEOUT)
return response.status_code == 200
except Exception as e:
logger.error(f"Telegram connection test failed: {e}")
return False
def send_message(self, text: str) -> bool:
"""Send message to Telegram chat"""
if not TELEGRAM_ENABLED or not self.bot_token or not self.chat_id:
logger.debug("Telegram notifications disabled or missing credentials")
return False
try:
url = f"{self.base_url}/sendMessage"
payload = {
'chat_id': self.chat_id,
'text': text,
'parse_mode': 'Markdown'
}
response = requests.post(url, json=payload, timeout=TELEGRAM_TIMEOUT)
result = response.json()
if result.get('ok'):
logger.info("Telegram notification sent successfully")
return True
else:
logger.error(f"Telegram API error: {result}")
return False
except Exception as e:
logger.error(f"Failed to send Telegram message: {e}")
return False
def format_position_message(self, last_closed: Optional[Dict], current_open: Dict) -> str:
"""Format position data into readable message"""
lines = ["NEW CLP POSITION DETECTED\n"]
# Previous closed position section
if last_closed:
lines.append("LAST CLOSED POSITION:")
lines.append(f"• Token ID: {last_closed.get('token_id', 'N/A')}")
lines.append(f"• Entry: ${last_closed.get('entry_price', 0):.2f}")
lines.append(f"• Target Value: ${last_closed.get('target_value', 0):.2f}")
# Add duration if timestamps available
if last_closed.get('timestamp_open') and last_closed.get('timestamp_close'):
duration = last_closed['timestamp_close'] - last_closed['timestamp_open']
hours = duration // 3600
minutes = (duration % 3600) // 60
lines.append(f"• Duration: {hours}h {minutes}m")
# Add hedge performance if available
hedge_pnl = last_closed.get('hedge_pnl_realized', 0)
if hedge_pnl != 0:
lines.append(f"• Hedge PnL: ${hedge_pnl:.2f}")
hedge_fees = last_closed.get('hedge_fees_paid', 0)
if hedge_fees != 0:
lines.append(f"• Hedge Fees: ${hedge_fees:.2f}")
else:
lines.append("LAST CLOSED POSITION: None")
lines.append("") # Empty line
# Current opened position section
lines.append("CURRENTLY OPENED:")
lines.append(f"• Token ID: {current_open.get('token_id', 'N/A')}")
lines.append(f"• Entry: ${current_open.get('entry_price', 0):.2f}")
lines.append(f"• Target Value: ${current_open.get('target_value', 0):.2f}")
# Range information
range_lower = current_open.get('range_lower', 0)
range_upper = current_open.get('range_upper', 0)
if range_lower and range_upper:
lines.append(f"• Range: ${range_lower:.2f} - ${range_upper:.2f}")
# Initial amounts
amount0 = current_open.get('amount0_initial', 0)
amount1 = current_open.get('amount1_initial', 0)
if amount0 and amount1:
lines.append(f"• Initial: {amount0:.4f} ETH + {amount1:.2f} USDC")
# Time since opening
if current_open.get('timestamp_open'):
age = int(time.time()) - current_open['timestamp_open']
hours = age // 3600
minutes = (age % 3600) // 60
lines.append(f"• Time: {hours}h {minutes}m ago")
# Performance comparison if we have both positions
if last_closed and current_open:
lines.append("") # Empty line
lines.append("PERFORMANCE COMPARISON:")
# Entry price change
last_entry = Decimal(str(last_closed.get('entry_price', 0)))
curr_entry = Decimal(str(current_open.get('entry_price', 0)))
if last_entry > 0:
entry_change = curr_entry - last_entry
entry_change_pct = (entry_change / last_entry) * 100
sign = "+" if entry_change >= 0 else ""
lines.append(f"• Entry Change: {sign}${entry_change:.2f} ({sign}{entry_change_pct:.2f}%)")
# Value change
last_value = Decimal(str(last_closed.get('target_value', 0)))
curr_value = Decimal(str(current_open.get('target_value', 0)))
if last_value > 0:
value_change = curr_value - last_value
value_change_pct = (value_change / last_value) * 100
sign = "+" if value_change >= 0 else ""
lines.append(f"• Value Change: {sign}${value_change:.2f} ({sign}{value_change_pct:.2f}%)")
# Hedge PnL trend
last_hedge_pnl = last_closed.get('hedge_pnl_realized', 0)
curr_hedge_equity = current_open.get('hedge_equity_usd', 0)
if last_hedge_pnl != 0 and curr_hedge_equity != 0:
hedge_trend = curr_hedge_equity - last_hedge_pnl
sign = "+" if hedge_trend >= 0 else ""
lines.append(f"• Hedge PnL Trend: ${last_hedge_pnl:.2f} -> ${curr_hedge_equity:.2f} ({sign}${hedge_trend:.2f})")
return "\n".join(lines)
class PositionMonitor:
"""Monitors hedge_status.json for changes"""
def __init__(self, json_file_path: str, state_file_path: str):
self.json_file_path = json_file_path
self.state_file_path = state_file_path
self.last_known_data = []
self.last_file_hash = ""
self.state = self.load_state()
def load_state(self) -> Dict[str, Any]:
"""Load monitor state from file"""
try:
if os.path.exists(self.state_file_path):
with open(self.state_file_path, 'r') as f:
return json.load(f)
except Exception as e:
logger.warning(f"Could not load state file: {e}")
return {
"last_known_open_positions": [],
"last_processed_timestamp": 0,
"last_file_hash": ""
}
def save_state(self):
"""Save monitor state to file"""
try:
with open(self.state_file_path, 'w') as f:
json.dump(self.state, f, indent=2)
except Exception as e:
logger.error(f"Could not save state file: {e}")
def get_file_hash(self, data: List[Dict]) -> str:
"""Generate hash of file content to detect changes"""
content = json.dumps(data, sort_keys=True)
return hashlib.md5(content.encode()).hexdigest()
def safe_read_json(self) -> List[Dict]:
"""Safely read JSON file with retry logic"""
attempts = 0
while attempts < 3:
try:
if not os.path.exists(self.json_file_path):
logger.warning(f"JSON file not found: {self.json_file_path}")
return []
with open(self.json_file_path, 'r') as f:
return json.load(f)
except (json.JSONDecodeError, IOError) as e:
logger.warning(f"Attempt {attempts + 1}: Error reading JSON file: {e}")
time.sleep(1)
attempts += 1
logger.error("Failed to read JSON file after 3 attempts")
return []
def extract_notification_data(self, data: List[Dict]) -> Tuple[Optional[Dict], Optional[Dict]]:
"""Extract last closed and current open positions"""
current_open = None
last_closed = None
# Find current open position
for item in data:
if item.get('status') == 'OPEN':
current_open = item
break
# Find most recent closed position
closed_positions = [item for item in data if item.get('status') == 'CLOSED']
if closed_positions:
# Sort by timestamp_open (descending) to get most recent
closed_positions.sort(key=lambda x: x.get('timestamp_open', 0), reverse=True)
last_closed = closed_positions[0]
return last_closed, current_open
def check_for_changes(self) -> bool:
"""Check if there are changes requiring notification"""
current_data = self.safe_read_json()
if not current_data:
return False
# Check if file content actually changed
current_hash = self.get_file_hash(current_data)
if current_hash == self.state.get("last_file_hash", ""):
return False
# Extract positions
last_closed, current_open = self.extract_notification_data(current_data)
if not current_open:
# No open position, nothing to notify about
return False
current_open_id = current_open.get('token_id')
last_known_opens = self.state.get("last_known_open_positions", [])
# Check if this is a new open position
if current_open_id not in last_known_opens:
# New position detected!
self.last_known_data = current_data
return True
return False
def get_notification_data(self) -> Tuple[Optional[Dict], Optional[Dict]]:
"""Get data for notification"""
current_data = self.safe_read_json()
return self.extract_notification_data(current_data)
def update_state(self):
"""Update internal state after notification"""
current_data = self.safe_read_json()
if current_data:
_, current_open = self.extract_notification_data(current_data)
if current_open:
# Update state with current open positions
self.state["last_known_open_positions"] = [current_open.get('token_id')]
self.state["last_processed_timestamp"] = int(time.time())
self.state["last_file_hash"] = self.get_file_hash(current_data)
self.save_state()
def main():
"""Main monitoring loop"""
logger.info("🤖 Telegram Monitor Starting...")
notifier = None
if not TELEGRAM_ENABLED:
logger.info("📵 Telegram notifications disabled (TELEGRAM_MONITOR_ENABLED=False)")
else:
if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID:
logger.error("❌ Telegram enabled but missing BOT_TOKEN or CHAT_ID")
return
# Initialize notifier and test connection
notifier = TelegramNotifier(TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID)
if not notifier.test_connection():
logger.error("❌ Telegram connection failed - check token and network")
return
logger.info(f"✅ Telegram connection established to chat ID: {TELEGRAM_CHAT_ID}")
# Initialize monitor
monitor = PositionMonitor(HEDGE_STATUS_FILE, TELEGRAM_STATE_FILE)
logger.info(f"Monitoring file: {HEDGE_STATUS_FILE}")
logger.info(f"Check interval: {TELEGRAM_CHECK_INTERVAL} seconds")
logger.info(f"State file: {TELEGRAM_STATE_FILE}")
try:
while True:
try:
if monitor.check_for_changes():
logger.info("New position opening detected!")
if TELEGRAM_ENABLED and notifier:
last_closed, current_open = monitor.get_notification_data()
if current_open:
message = notifier.format_position_message(last_closed, current_open)
success = notifier.send_message(message)
if success:
monitor.update_state()
logger.info(f"Notification sent for position {current_open.get('token_id')}")
else:
logger.error("Failed to send notification")
else:
logger.warning("Position change detected but no open position found")
else:
logger.info("Telegram disabled or notifier not available - skipping notification")
monitor.update_state() # Still update state to avoid loops
else:
logger.debug("No changes detected")
time.sleep(TELEGRAM_CHECK_INTERVAL)
except KeyboardInterrupt:
break
except Exception as e:
logger.error(f"Error in monitoring loop: {e}")
time.sleep(TELEGRAM_CHECK_INTERVAL) # Continue after error
except KeyboardInterrupt:
logger.info("Shutting down Telegram Monitor...")
logger.info("Telegram Monitor stopped")
if __name__ == "__main__":
main()

109
tools/telegram_setup.py Normal file
View File

@ -0,0 +1,109 @@
#!/usr/bin/env python3
"""
Setup script for Telegram Bot notifications
Helps users create a Telegram bot and get required credentials
"""
import webbrowser
import time
def print_instructions():
"""Print setup instructions"""
print("🤖 TELEGRAM BOT SETUP INSTRUCTIONS")
print("=" * 50)
print()
print("1️⃣ CREATE YOUR TELEGRAM BOT:")
print(" • Open Telegram and search for @BotFather")
print(" • Send: /newbot")
print(" • Choose a name for your bot")
print(" • Choose a username (must end with 'bot')")
print(" • BotFather will give you a HTTP API token")
print(" • Copy this token (it looks like: 123456789:ABCdefGHIjklMNOpqrsTUVwxyz)")
print()
print("2️⃣ GET YOUR CHAT ID:")
print(" • Send a message to your new bot first (any message)")
print(" • Your bot will not work if you haven't messaged it first")
print(" • Use the script below to get your Chat ID")
print()
print("3️⃣ CONFIGURE ENVIRONMENT:")
print(" • Add to your .env file:")
print(" TELEGRAM_MONITOR_ENABLED=True")
print(" TELEGRAM_BOT_TOKEN=your_bot_token_here")
print(" TELEGRAM_CHAT_ID=your_chat_id_here")
print()
print("4️⃣ RUN THE MONITOR:")
print(" • python tools/telegram_monitor.py")
print()
print("=" * 50)
def get_chat_id():
"""Get chat ID using bot token"""
print("🔍 CHAT ID FINDER")
print("-" * 20)
token = input("Enter your bot token: ").strip()
if not token:
print("❌ No token provided")
return
try:
import requests
import json
url = f"https://api.telegram.org/bot{token}/getUpdates"
response = requests.get(url, timeout=10)
data = response.json()
if data.get('ok'):
updates = data.get('result', [])
if updates:
chat_id = updates[-1]['message']['chat']['id']
print(f"✅ Your Chat ID: {chat_id}")
print()
print("Add this to your .env file:")
print(f"TELEGRAM_CHAT_ID={chat_id}")
else:
print("❌ No messages found")
print("Send a message to your bot first, then try again")
else:
print(f"❌ Error: {data.get('description', 'Unknown error')}")
except Exception as e:
print(f"❌ Error getting Chat ID: {e}")
def main():
"""Main setup menu"""
print("🤖 CLP TELEGRAM NOTIFICATION SETUP")
print("=" * 40)
print()
while True:
print("Choose an option:")
print("1. Show setup instructions")
print("2. Get Chat ID")
print("3. Open BotFather (to create bot)")
print("4. Exit")
print()
choice = input("Enter choice (1-4): ").strip()
if choice == '1':
print_instructions()
elif choice == '2':
get_chat_id()
elif choice == '3':
print("Opening BotFather in Telegram...")
try:
webbrowser.open('https://t.me/BotFather')
except:
print("Could not open browser. Go to https://t.me/BotFather manually")
elif choice == '4':
print("Goodbye! 🚀")
break
else:
print("Invalid choice. Try again.")
print("\n" + "=" * 40 + "\n")
if __name__ == "__main__":
main()

105
tools/test_telegram_monitor.py Normal file
View File

@ -0,0 +1,105 @@
#!/usr/bin/env python3
"""
Test script for Telegram Monitor functionality
Validates JSON parsing and message formatting
"""
import sys
import os
# Add the tools directory to the path
current_dir = os.path.dirname(os.path.abspath(__file__))
sys.path.append(current_dir)
from telegram_monitor import TelegramNotifier, PositionMonitor
def test_json_parsing():
"""Test JSON parsing from hedge_status.json"""
print("Testing JSON parsing...")
monitor = PositionMonitor("../hedge_status.json", "test_state.json")
data = monitor.safe_read_json()
if not data:
print("❌ Failed to read JSON data")
return False
print(f"✅ Successfully read {len(data)} positions")
# Find open and closed positions
last_closed, current_open = monitor.extract_notification_data(data)
if current_open:
print(f"✅ Found open position: Token ID {current_open.get('token_id')}")
else:
print("ℹ️ No open position found")
if last_closed:
print(f"✅ Found last closed: Token ID {last_closed.get('token_id')}")
else:
print("ℹ️ No closed positions found")
return True
def test_message_formatting():
"""Test message formatting"""
print("\n🧪 Testing message formatting...")
# Create test data
last_closed = {
"token_id": 1234567,
"entry_price": 3000.0,
"target_value": 2000.0,
"hedge_pnl_realized": -15.50,
"hedge_fees_paid": 2.25,
"timestamp_open": 1766328197,
"timestamp_close": 1766331797
}
current_open = {
"token_id": 1234568,
"entry_price": 2980.0,
"target_value": 1950.0,
"range_lower": 2900.0,
"range_upper": 3060.0,
"amount0_initial": 0.3,
"amount1_initial": 900.0,
"timestamp_open": 1766335400
}
notifier = TelegramNotifier("test_token", "test_chat")
message = notifier.format_position_message(last_closed, current_open)
print("📱 Sample message:")
print("-" * 40)
print(message)
print("-" * 40)
return True
def main():
print("Telegram Monitor Test Suite")
print("=" * 50)
success = True
try:
if not test_json_parsing():
success = False
if not test_message_formatting():
success = False
except Exception as e:
print(f"❌ Test failed with error: {e}")
success = False
if success:
print("\n✅ All tests passed! Telegram monitor is ready.")
else:
print("\n❌ Some tests failed. Check the errors above.")
return 0 if success else 1
if __name__ == "__main__":
sys.exit(main())

64
tools/test_telegram_simple.py Normal file
View File

@ -0,0 +1,64 @@
#!/usr/bin/env python3
"""
Simple test for Telegram Monitor functionality
"""
import sys
import os
# Add tools directory to path
current_dir = os.path.dirname(os.path.abspath(__file__))
sys.path.append(current_dir)
try:
from telegram_monitor import TelegramNotifier, PositionMonitor
print("Testing Telegram Monitor components...")
# Test with sample data
notifier = TelegramNotifier("test_token", "test_chat")
last_closed = {
"token_id": 1234567,
"entry_price": 3000.0,
"target_value": 2000.0,
"hedge_pnl_realized": -15.50,
"hedge_fees_paid": 2.25,
"timestamp_open": 1766328197,
"timestamp_close": 1766331797
}
current_open = {
"token_id": 1234568,
"entry_price": 2980.0,
"target_value": 1950.0,
"range_lower": 2900.0,
"range_upper": 3060.0,
"amount0_initial": 0.3,
"amount1_initial": 900.0,
"timestamp_open": 1766335400
}
message = notifier.format_position_message(last_closed, current_open)
print("Message formatting test PASSED")
print("Sample message:")
print("-" * 40)
print(message)
print("-" * 40)
# Test JSON reading if file exists
if os.path.exists("../hedge_status.json"):
monitor = PositionMonitor("../hedge_status.json", "test_state.json")
data = monitor.safe_read_json()
if data:
print(f"JSON reading test PASSED - found {len(data)} positions")
else:
print("JSON reading test FAILED")
print("All tests completed successfully!")
except ImportError as e:
print(f"Import error: {e}")
print("Make sure telegram_monitor.py is in the same directory")
except Exception as e:
print(f"Test error: {e}")