diff --git a/price_feeds/python-rest-api/README.md b/price_feeds/python-rest-api/README.md new file mode 100644 index 00000000..8437b4a4 --- /dev/null +++ b/price_feeds/python-rest-api/README.md @@ -0,0 +1,231 @@ +# Python REST API for Pyth Price Feeds + +A Flask-based REST API for monitoring cryptocurrency prices using Pyth Network's decentralized oracle. This example demonstrates how to build a production-ready API with Pyth price feeds, supporting 20+ tokens including BTC, ETH, SOL, and more. + +## Features + +- ✅ **20+ Cryptocurrencies** - Monitor BTC, ETH, SOL, BNB, AVAX, MATIC, ARB, OP, and more +- ✅ **Threshold Monitoring** - Check if prices cross specific thresholds +- ✅ **Real-time Prices** - Powered by Pyth Network's Hermes API +- ✅ **Session Management** - Monitor multiple tokens simultaneously +- ✅ **RESTful API** - Clean, well-documented endpoints +- ✅ **CORS Enabled** - Ready for web applications + +## Quick Start + +### 1. Install Dependencies + +```bash +pip install -r requirements.txt +``` + +### 2. Start the Server + +```bash +python app.py +``` + +Server will start on `http://localhost:8080` + +### 3. Test the API + +```bash +# In another terminal +python test_endpoints.py +``` + +## API Endpoints + +| Endpoint | Method | Description | +|----------|--------|-------------| +| `/health` | GET | Health check | +| `/api/tokens` | GET | List all supported tokens | +| `/api/price/` | GET | Get current price for any token | +| `/api/check` | POST | Check price vs threshold (single) | +| `/api/monitor/start` | POST | Start monitoring session | +| `/api/monitor/` | GET | Get session status | +| `/api/monitor//stop` | POST | Stop monitoring session | +| `/api/monitor/sessions` | GET | List all sessions | + +## Usage Examples + +### Get Bitcoin Price + +```bash +curl http://localhost:8080/api/price/BTC/USD +``` + +**Response:** +```json +{ + "success": true, + "symbol": "BTC/USD", + "price": 67234.56, + "timestamp": "2024-01-23T10:15:23.123456" +} +``` + +### Check if ETH is Below $3000 + +```bash +curl -X POST http://localhost:8080/api/check \ + -H "Content-Type: application/json" \ + -d '{"symbol": "ETH/USD", "threshold": 3000}' +``` + +**Response:** +```json +{ + "success": true, + "symbol": "ETH/USD", + "price": 2845.67, + "threshold": 3000, + "is_below_threshold": true, + "result": true, + "timestamp": "2024-01-23T10:15:23.123456" +} +``` + +### Start Continuous Monitoring + +```bash +curl -X POST http://localhost:8080/api/monitor/start \ + -H "Content-Type: application/json" \ + -d '{ + "symbol": "BTC/USD", + "threshold": 50000, + "update_interval": 10 + }' +``` + +**Response:** +```json +{ + "success": true, + "message": "Started monitoring BTC/USD with threshold $50000.00", + "session_id": "session_1", + "symbol": "BTC/USD", + "threshold": 50000, + "update_interval": 10 +} +``` + +## Supported Tokens + +BTC/USD, ETH/USD, SOL/USD, BNB/USD, AVAX/USD, MATIC/USD, ARB/USD, OP/USD, DOGE/USD, ADA/USD, DOT/USD, LINK/USD, UNI/USD, ATOM/USD, XRP/USD, LTC/USD, APT/USD, SUI/USD, TRX/USD, NEAR/USD + +Get the complete list: +```bash +curl http://localhost:8080/api/tokens +``` + +## Project Structure + +``` +python-rest-api/ +├── app.py # Flask REST API server +├── price_monitor.py # Core price monitoring logic +├── requirements.txt # Python dependencies +├── test_endpoints.py # API test suite +└── README.md # This file +``` + +## Key Components + +### price_monitor.py + +Core module that handles Pyth Network integration: +- Fetches prices from Pyth's Hermes API +- Supports 20+ cryptocurrency pairs +- Provides both single-check and continuous monitoring +- Handles price threshold comparisons + +### app.py + +Flask REST API server that provides: +- RESTful endpoints for price queries +- Session-based continuous monitoring +- Background threading for real-time updates +- CORS support for web applications + +## Python Client Example + +```python +import requests +import time + +BASE_URL = "http://localhost:8080" + +# Get current price +response = requests.get(f"{BASE_URL}/api/price/BTC/USD") +price_data = response.json() +print(f"BTC Price: ${price_data['price']:,.2f}") + +# Quick threshold check +response = requests.post(f"{BASE_URL}/api/check", json={ + "symbol": "ETH/USD", + "threshold": 3000 +}) +result = response.json() +print(f"ETH below $3000: {result['is_below_threshold']}") + +# Start monitoring +response = requests.post(f"{BASE_URL}/api/monitor/start", json={ + "symbol": "BTC/USD", + "threshold": 50000, + "update_interval": 10 +}) +session_id = response.json()['session_id'] +print(f"Started monitoring, session: {session_id}") + +# Poll for updates +try: + while True: + response = requests.get(f"{BASE_URL}/api/monitor/{session_id}") + data = response.json()['data'] + + if data['price']: + print(f"[{data['timestamp']}] {data['symbol']}: " + f"${data['price']:,.2f} | Below threshold: {data['is_below_threshold']}") + + time.sleep(10) +except KeyboardInterrupt: + # Stop monitoring + requests.post(f"{BASE_URL}/api/monitor/{session_id}/stop") + print("\nMonitoring stopped") +``` + +## Use Cases + +- **Price Alerts** - Build notification systems for price thresholds +- **Trading Bots** - Integrate real-time price data for automated trading +- **Portfolio Tracking** - Monitor multiple assets simultaneously +- **Market Analysis** - Collect and analyze price trends +- **DeFi Integrations** - Use as a data feed for decentralized applications + +## Technologies + +- **Flask** - Lightweight web framework +- **Pyth Network** - Decentralized oracle for real-time price data +- **Python 3.7+** - Programming language +- **Threading** - For concurrent monitoring sessions + +## Deployment + +This API can be easily deployed using: +- **ngrok** - For quick public exposure (see deployment guide) +- **Docker** - Containerized deployment +- **Cloud Services** - AWS, GCP, Azure, Heroku, etc. + +## About Pyth Network + +This example uses [Pyth Network](https://pyth.network)'s Hermes API to fetch real-time cryptocurrency prices. Pyth Network is a decentralized oracle that provides high-fidelity, high-frequency market data from over 90+ first-party publishers. + +## License + +Apache 2.0 + +--- + +**Built with [Pyth Network](https://pyth.network) 🔮** + diff --git a/price_feeds/python-rest-api/app.py b/price_feeds/python-rest-api/app.py new file mode 100644 index 00000000..58640a89 --- /dev/null +++ b/price_feeds/python-rest-api/app.py @@ -0,0 +1,364 @@ +#!/usr/bin/env python3 +""" +Flask API for Multi-Token Price Monitor using Pyth Network +Supports monitoring any cryptocurrency available on Pyth Network +""" + +from flask import Flask, jsonify, request +from flask_cors import CORS +from price_monitor import PriceMonitor +import threading +import time +from datetime import datetime + +app = Flask(__name__) +CORS(app) # Enable CORS for all routes + +# Global state for monitoring sessions +monitoring_sessions = {} # {session_id: {monitor, thread, is_running, latest_data}} +session_counter = 0 + + +@app.route('/health', methods=['GET']) +def health_check(): + """Health check endpoint""" + return jsonify({ + "status": "healthy", + "service": "crypto-price-monitor", + "timestamp": datetime.now().isoformat() + }) + + +@app.route('/api/tokens', methods=['GET']) +def get_available_tokens(): + """Get list of all available tokens""" + tokens = PriceMonitor.get_available_tokens() + print('hi') + return jsonify({ + "success": True, + "count": len(tokens), + "tokens": tokens + }) + + + +@app.route('/api/price/', methods=['GET']) +def get_token_price(symbol): + """ + Get current price for any token + + Path parameter: + symbol: Token symbol (e.g., BTC/USD, ETH/USD) + """ + symbol = symbol.upper() + print('hi') + + + # Check if symbol is supported + if symbol not in PriceMonitor.get_available_tokens(): + print('hi') + + return jsonify({ + "success": False, + "error": f"Unsupported token: {symbol}", + "available_tokens": PriceMonitor.get_available_tokens() + }), 400 + + # Fetch price + price = PriceMonitor.get_price_for_symbol(symbol) + + if price is not None: + print('hi') + + return jsonify({ + "success": True, + "symbol": symbol, + "price": price, + "timestamp": datetime.now().isoformat() + }) + + else: + print('hi') + + return jsonify({ + "success": False, + "error": f"Failed to fetch price for {symbol}" + }), 500 + + +@app.route('/api/check', methods=['POST']) +def check_price_threshold(): + """ + Check if a token's current price is below a threshold (single check) + + Request body: + { + "symbol": "BTC/USD", + "threshold": 50000 + } + """ + data = request.get_json() + + if not data: + return jsonify({ + "success": False, + "error": "Request body is required" + }), 400 + + symbol = data.get('symbol', '').upper() + threshold = data.get('threshold') + + if not symbol or threshold is None: + return jsonify({ + "success": False, + "error": "Both 'symbol' and 'threshold' are required" + }), 400 + + try: + threshold = float(threshold) + except (ValueError, TypeError): + return jsonify({ + "success": False, + "error": "Threshold must be a number" + }), 400 + + # Check if symbol is supported + if symbol not in PriceMonitor.get_available_tokens(): + return jsonify({ + "success": False, + "error": f"Unsupported token: {symbol}", + "available_tokens": PriceMonitor.get_available_tokens() + }), 400 + + # Create temporary monitor and check + try: + monitor = PriceMonitor(symbol, threshold) + price, is_below = monitor.get_single_check() + + if price is not None: + return jsonify({ + "success": True, + "symbol": symbol, + "price": price, + "threshold": threshold, + "is_below_threshold": is_below, + "result": is_below, # Boolean result + "timestamp": datetime.now().isoformat() + }) + else: + return jsonify({ + "success": False, + "error": f"Failed to fetch price for {symbol}" + }), 500 + + except ValueError as e: + return jsonify({ + "success": False, + "error": str(e) + }), 400 + + +def background_monitor(session_id): + """Background thread that continuously monitors token price""" + session = monitoring_sessions[session_id] + monitor = session['monitor'] + + while session['is_running']: + try: + price = monitor.get_price() + if price is not None: + is_below = monitor.check_threshold(price) + session['latest_data'] = { + "symbol": monitor.symbol, + "price": price, + "is_below_threshold": is_below, + "threshold": monitor.threshold, + "timestamp": datetime.now().isoformat(), + "status": "Monitoring" + } + time.sleep(monitor.update_interval) + except Exception as e: + print(f"Error in monitoring thread for session {session_id}: {e}") + time.sleep(5) + + +@app.route('/api/monitor/start', methods=['POST']) +def start_monitoring(): + """ + Start monitoring a token's price against a threshold + + Request body: + { + "symbol": "BTC/USD", + "threshold": 50000, + "update_interval": 10 # optional, defaults to 10 + } + + Returns a session_id to track this monitoring session + """ + global session_counter + + data = request.get_json() + + if not data: + return jsonify({ + "success": False, + "error": "Request body is required" + }), 400 + + symbol = data.get('symbol', '').upper() + threshold = data.get('threshold') + update_interval = data.get('update_interval', 10.0) + + if not symbol or threshold is None: + return jsonify({ + "success": False, + "error": "Both 'symbol' and 'threshold' are required" + }), 400 + + try: + threshold = float(threshold) + update_interval = float(update_interval) + except (ValueError, TypeError): + return jsonify({ + "success": False, + "error": "Threshold and update_interval must be numbers" + }), 400 + + # Check if symbol is supported + if symbol not in PriceMonitor.get_available_tokens(): + return jsonify({ + "success": False, + "error": f"Unsupported token: {symbol}", + "available_tokens": PriceMonitor.get_available_tokens() + }), 400 + + # Create monitor + try: + monitor = PriceMonitor(symbol, threshold, update_interval) + except ValueError as e: + return jsonify({ + "success": False, + "error": str(e) + }), 400 + + # Create new session + session_counter += 1 + session_id = f"session_{session_counter}" + + monitoring_sessions[session_id] = { + 'monitor': monitor, + 'is_running': True, + 'latest_data': { + "symbol": symbol, + "price": None, + "is_below_threshold": None, + "threshold": threshold, + "timestamp": None, + "status": "Starting..." + } + } + + # Start background monitoring thread + thread = threading.Thread(target=background_monitor, args=(session_id,), daemon=True) + monitoring_sessions[session_id]['thread'] = thread + thread.start() + + return jsonify({ + "success": True, + "message": f"Started monitoring {symbol} with threshold ${threshold:.2f}", + "session_id": session_id, + "symbol": symbol, + "threshold": threshold, + "update_interval": update_interval + }) + + +@app.route('/api/monitor/', methods=['GET']) +def get_monitor_status(session_id): + """ + Get current status of a monitoring session + + Path parameter: + session_id: The session ID returned from /api/monitor/start + """ + if session_id not in monitoring_sessions: + return jsonify({ + "success": False, + "error": "Session not found" + }), 404 + + session = monitoring_sessions[session_id] + return jsonify({ + "success": True, + "session_id": session_id, + "is_running": session['is_running'], + "data": session['latest_data'] + }) + + +@app.route('/api/monitor//stop', methods=['POST']) +def stop_monitoring(session_id): + """ + Stop a monitoring session + + Path parameter: + session_id: The session ID to stop + """ + if session_id not in monitoring_sessions: + return jsonify({ + "success": False, + "error": "Session not found" + }), 404 + + session = monitoring_sessions[session_id] + session['is_running'] = False + session['latest_data']['status'] = "Stopped" + + return jsonify({ + "success": True, + "message": f"Stopped monitoring session {session_id}" + }) + + +@app.route('/api/monitor/sessions', methods=['GET']) +def list_monitoring_sessions(): + """Get list of all active monitoring sessions""" + sessions = [] + for session_id, session in monitoring_sessions.items(): + sessions.append({ + "session_id": session_id, + "symbol": session['monitor'].symbol, + "threshold": session['monitor'].threshold, + "is_running": session['is_running'], + "status": session['latest_data']['status'] + }) + + return jsonify({ + "success": True, + "count": len(sessions), + "sessions": sessions + }) + + +if __name__ == '__main__': + print("=" * 70) + print("Multi-Token Price Monitor API - Powered by Pyth Network") + print("=" * 70) + print("\nStarting Flask server...") + print("Local access: http://localhost:5000") + print("\nTo expose publicly with ngrok:") + print(" ngrok http 5000") + print("\nAPI Endpoints:") + print(" GET /health - Health check") + print(" GET /api/tokens - List all available tokens") + print(" GET /api/price/ - Get current price for any token") + print(" POST /api/check - Check price vs threshold (single)") + print(" POST /api/monitor/start - Start monitoring session") + print(" GET /api/monitor/ - Get monitoring session status") + print(" POST /api/monitor//stop - Stop monitoring session") + print(" GET /api/monitor/sessions - List all monitoring sessions") + print("=" * 70) + print() + + app.run(host='0.0.0.0', port=8080, debug=False) diff --git a/price_feeds/python-rest-api/price_monitor.py b/price_feeds/python-rest-api/price_monitor.py new file mode 100755 index 00000000..fd27e71d --- /dev/null +++ b/price_feeds/python-rest-api/price_monitor.py @@ -0,0 +1,252 @@ +#!/usr/bin/env python3 +""" +Multi-Token Price Monitor using Pyth Network +Monitors any token price and checks if it's below a given threshold +""" + +import requests +import time +from datetime import datetime +from typing import Optional, Dict + +class PriceMonitor: + """ + Monitor cryptocurrency prices using Pyth Network's price feeds + """ + + # Pyth Network Hermes API endpoint + BASE_URL = "https://hermes.pyth.network" + + # Common token price feed IDs from Pyth Network + FEED_IDS = { + # Crypto/USD pairs + "BTC/USD": "0xe62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43", + "ETH/USD": "0xff61491a931112ddf1bd8147cd1b641375f79f5825126d665480874634fd0ace", + "SOL/USD": "0xef0d8b6fda2ceba41da15d4095d1da392a0d2f8ed0c6c7bc0f4cfac8c280b56d", + "BNB/USD": "0x2f95862b045670cd22bee3114c39763a4a08beeb663b145d283c31d7d1101c4f", + "AVAX/USD": "0x93da3352f9f1d105fdfe4971cfa80e9dd777bfc5d0f683ebb6e1294b92137bb7", + "MATIC/USD": "0x5de33a9112c2b700b8d30b8a3402c103578ccfa2765696471cc672bd5cf6ac52", + "ARB/USD": "0x3fa4252848f9f0a1480be62745a4629d9eb1322aebab8a791e344b3b9c1adcf5", + "OP/USD": "0x385f64d993f7b77d8182ed5003d97c60aa3361f3cecfe711544d2d59165e9bdf", + "DOGE/USD": "0xdcef50dd0a4cd2dcc17e45df1676dcb336a11a61c69df7a0299b0150c672d25c", + "ADA/USD": "0x2a01deaec9e51a579277b34b122399984d0bbf57e2458a7e42fecd2829867a0d", + "DOT/USD": "0xca3eed9b267293f6595901c734c7525ce8ef49adafe8284606ceb307afa2ca5b", + "LINK/USD": "0x8ac0c70fff57e9aefdf5edf44b51d62c2d433653cbb2cf5cc06bb115af04d221", + "UNI/USD": "0x78d185a741d07edb3412b09008b7c5cfb9bbbd7d568bf00ba737b456ba171501", + "ATOM/USD": "0xb00b60f88b03a6a625a8d1c048c3f66653edf217439983d037e7222c4e612819", + "XRP/USD": "0xec5d399846a9209f3fe5881d70aae9268c94339ff9817e8d18ff19fa05eea1c8", + "LTC/USD": "0x6e3f3fa8253588df9326580180233eb791e03b443a3ba7a1d892e73874e19a54", + "APT/USD": "0x03ae4db29ed4ae33d323568895aa00337e658e348b37509f5372ae51f0af00d5", + "SUI/USD": "0x23d7315113f5b1d3ba7a83604c44b94d79f4fd69af77f804fc7f920a6dc65744", + "TRX/USD": "0x67aed5a24fdad045475e7195c98a98aea119c763f272d4523f5bac93a4f33c2b", + "NEAR/USD": "0xc415de8d2eba7db216527dff4b60e8f3a5311c740dadb233e13e12547e226750", + } + + def __init__(self, symbol: str, threshold: float, update_interval: float = 10.0): + """ + Initialize the price monitor + + Args: + symbol: Token symbol (e.g., "BTC/USD", "ETH/USD") + threshold: Price threshold in USD + update_interval: Time in seconds between price checks (default 10 seconds) + """ + self.symbol = symbol.upper() + self.threshold = threshold + self.update_interval = update_interval + self.is_running = False + + # Get feed ID for the symbol + self.feed_id = self.FEED_IDS.get(self.symbol) + if not self.feed_id: + raise ValueError(f"Unsupported token symbol: {symbol}. Available: {', '.join(self.FEED_IDS.keys())}") + + @classmethod + def get_available_tokens(cls) -> list: + """Get list of available token symbols""" + return list(cls.FEED_IDS.keys()) + + @classmethod + def get_feed_id_for_symbol(cls, symbol: str) -> Optional[str]: + """Get Pyth feed ID for a given symbol""" + return cls.FEED_IDS.get(symbol.upper()) + + def get_price(self, feed_id: Optional[str] = None) -> Optional[float]: + """ + Fetch the current price from Pyth Network + + Args: + feed_id: Optional feed ID to fetch. If None, uses self.feed_id + + Returns: + Current price in USD, or None if request fails + """ + target_feed_id = feed_id or self.feed_id + + try: + # Use Pyth's latest price feeds endpoint + url = f"{self.BASE_URL}/v2/updates/price/latest" + params = { + "ids[]": target_feed_id + } + + response = requests.get(url, params=params, timeout=10) + response.raise_for_status() + + data = response.json() + + # Parse the price data + if "parsed" in data and len(data["parsed"]) > 0: + price_data = data["parsed"][0]["price"] + price = float(price_data["price"]) + expo = int(price_data["expo"]) + + # Calculate actual price (price * 10^expo) + actual_price = price * (10 ** expo) + return actual_price + + return None + + except requests.exceptions.RequestException as e: + print(f"Error fetching price: {e}") + return None + except (KeyError, ValueError, IndexError) as e: + print(f"Error parsing price data: {e}") + return None + + @classmethod + def get_price_for_symbol(cls, symbol: str) -> Optional[float]: + """ + Get current price for any supported symbol without creating a monitor + + Args: + symbol: Token symbol (e.g., "BTC/USD") + + Returns: + Current price in USD, or None if request fails + """ + feed_id = cls.get_feed_id_for_symbol(symbol) + if not feed_id: + return None + + try: + url = f"{cls.BASE_URL}/v2/updates/price/latest" + params = {"ids[]": feed_id} + + response = requests.get(url, params=params, timeout=10) + response.raise_for_status() + + data = response.json() + if "parsed" in data and len(data["parsed"]) > 0: + price_data = data["parsed"][0]["price"] + price = float(price_data["price"]) + expo = int(price_data["expo"]) + return price * (10 ** expo) + + return None + except Exception as e: + print(f"Error fetching price for {symbol}: {e}") + return None + + def check_threshold(self, current_price: float) -> bool: + """ + Check if current price is below threshold + + Args: + current_price: Current price in USD + + Returns: + True if price is below threshold, False otherwise + """ + return current_price < self.threshold + + def get_single_check(self) -> tuple[Optional[float], Optional[bool]]: + """ + Perform a single price check + + Returns: + Tuple of (current_price, is_below_threshold) + """ + current_price = self.get_price() + if current_price is not None: + is_below = self.check_threshold(current_price) + return current_price, is_below + return None, None + + +def main(): + """ + Main function to run the price monitor + """ + print("=" * 60) + print("Crypto Price Monitor - Powered by Pyth Network") + print("=" * 60) + print() + print("Available tokens:") + for i, symbol in enumerate(PriceMonitor.get_available_tokens(), 1): + print(f" {i}. {symbol}") + print() + + try: + # Get symbol from user + symbol_input = input("Enter token symbol (e.g., BTC/USD, ETH/USD): ").strip() + + # Get threshold from user + threshold_input = input("Enter price threshold in USD (e.g., 3000): ") + threshold = float(threshold_input) + + if threshold <= 0: + print("Error: Threshold must be a positive number") + return + + # Get update interval from user (optional) + interval_input = input("Enter update interval in seconds (default 10): ").strip() + update_interval = float(interval_input) if interval_input else 10.0 + + if update_interval < 0.5: + print("Warning: Setting interval to minimum of 0.5 seconds to avoid rate limiting") + update_interval = 0.5 + + print() + + # Create monitor + monitor = PriceMonitor(symbol_input, threshold, update_interval) + + print(f"Starting {monitor.symbol} price monitor...") + print(f"Threshold: ${threshold:.2f}") + print(f"Update interval: {update_interval} seconds") + print("-" * 60) + + # Continuous monitoring + monitor.is_running = True + try: + while monitor.is_running: + current_price = monitor.get_price() + + if current_price is not None: + is_below = monitor.check_threshold(current_price) + timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + status = "BELOW" if is_below else "ABOVE" + color = "\033[91m" if is_below else "\033[92m" # Red if below, Green if above + reset = "\033[0m" + + print(f"[{timestamp}] {monitor.symbol}: ${current_price:,.2f} | " + f"Status: {color}{status}{reset} threshold (${threshold:.2f}) | " + f"Result: {is_below}") + + time.sleep(update_interval) + + except KeyboardInterrupt: + print("\n\nMonitoring stopped by user") + monitor.is_running = False + + except ValueError as e: + print(f"Error: {e}") + except KeyboardInterrupt: + print("\n\nExiting...") + + +if __name__ == "__main__": + main() + diff --git a/price_feeds/python-rest-api/requirements.txt b/price_feeds/python-rest-api/requirements.txt new file mode 100644 index 00000000..e304cc59 --- /dev/null +++ b/price_feeds/python-rest-api/requirements.txt @@ -0,0 +1,4 @@ +requests==2.31.0 +flask==3.0.0 +flask-cors==4.0.0 + diff --git a/price_feeds/python-rest-api/test_endpoints.py b/price_feeds/python-rest-api/test_endpoints.py new file mode 100755 index 00000000..443f5afe --- /dev/null +++ b/price_feeds/python-rest-api/test_endpoints.py @@ -0,0 +1,262 @@ +#!/usr/bin/env python3 +""" +Test script for Multi-Token Price Monitor API +Tests all endpoints to verify functionality +""" + +import requests +import time +import sys + +BASE_URL = "http://localhost:5000" + + +def test_health(): + """Test health check endpoint""" + print("\n" + "="*60) + print("Test 1: Health Check") + print("="*60) + + try: + response = requests.get(f"{BASE_URL}/health", timeout=5) + if response.status_code == 200: + print("✅ PASSED - API is healthy") + print(f" Response: {response.json()}") + return True + else: + print(f"❌ FAILED - Status code: {response.status_code}") + return False + except Exception as e: + print(f"❌ FAILED - Error: {e}") + print("\n⚠️ Make sure the server is running: python app.py") + return False + + +def test_get_tokens(): + """Test get available tokens endpoint""" + print("\n" + "="*60) + print("Test 2: Get Available Tokens") + print("="*60) + + try: + response = requests.get(f"{BASE_URL}/api/tokens", timeout=5) + if response.status_code == 200: + data = response.json() + print(f"✅ PASSED - Found {data['count']} tokens") + print(f" Sample tokens: {', '.join(data['tokens'][:5])}...") + return True + else: + print(f"❌ FAILED - Status code: {response.status_code}") + return False + except Exception as e: + print(f"❌ FAILED - Error: {e}") + return False + + +def test_get_price(): + """Test get token price endpoint""" + print("\n" + "="*60) + print("Test 3: Get Token Price") + print("="*60) + + symbols = ["BTC/USD", "ETH/USD", "SOL/USD"] + + for symbol in symbols: + try: + response = requests.get(f"{BASE_URL}/api/price/{symbol}", timeout=10) + if response.status_code == 200: + data = response.json() + print(f"✅ {symbol}: ${data['price']:,.2f}") + else: + print(f"❌ FAILED for {symbol} - Status: {response.status_code}") + return False + except Exception as e: + print(f"❌ FAILED for {symbol} - Error: {e}") + return False + + return True + + +def test_check_threshold(): + """Test check price threshold endpoint""" + print("\n" + "="*60) + print("Test 4: Check Price vs Threshold") + print("="*60) + + test_cases = [ + {"symbol": "BTC/USD", "threshold": 50000}, + {"symbol": "ETH/USD", "threshold": 3000}, + {"symbol": "SOL/USD", "threshold": 100}, + ] + + for test in test_cases: + try: + response = requests.post( + f"{BASE_URL}/api/check", + json=test, + timeout=10 + ) + if response.status_code == 200: + data = response.json() + status = "BELOW" if data['is_below_threshold'] else "ABOVE" + print(f"✅ {test['symbol']}: ${data['price']:,.2f} is {status} ${test['threshold']:,.2f}") + else: + print(f"❌ FAILED for {test['symbol']} - Status: {response.status_code}") + return False + except Exception as e: + print(f"❌ FAILED for {test['symbol']} - Error: {e}") + return False + + return True + + +def test_monitoring(): + """Test monitoring session endpoints""" + print("\n" + "="*60) + print("Test 5: Monitoring Sessions") + print("="*60) + + # Start monitoring + print("\n📍 Starting monitoring session...") + try: + response = requests.post( + f"{BASE_URL}/api/monitor/start", + json={ + "symbol": "BTC/USD", + "threshold": 50000, + "update_interval": 5 + }, + timeout=5 + ) + + if response.status_code != 200: + print(f"❌ FAILED to start monitoring - Status: {response.status_code}") + return False + + data = response.json() + session_id = data['session_id'] + print(f"✅ Started monitoring session: {session_id}") + print(f" Symbol: {data['symbol']}") + print(f" Threshold: ${data['threshold']:,.2f}") + + except Exception as e: + print(f"❌ FAILED to start monitoring - Error: {e}") + return False + + # Wait for first update + print("\n⏳ Waiting 6 seconds for first update...") + time.sleep(6) + + # Check status + print("\n📍 Checking session status...") + try: + response = requests.get(f"{BASE_URL}/api/monitor/{session_id}", timeout=5) + + if response.status_code == 200: + data = response.json() + session_data = data['data'] + + if session_data['price']: + print(f"✅ Got monitoring data:") + print(f" Price: ${session_data['price']:,.2f}") + print(f" Below threshold: {session_data['is_below_threshold']}") + print(f" Status: {session_data['status']}") + else: + print("⚠️ Monitoring started but no data yet (needs more time)") + else: + print(f"❌ FAILED to get status - Status: {response.status_code}") + return False + + except Exception as e: + print(f"❌ FAILED to get status - Error: {e}") + return False + + # List sessions + print("\n📍 Listing all sessions...") + try: + response = requests.get(f"{BASE_URL}/api/monitor/sessions", timeout=5) + + if response.status_code == 200: + data = response.json() + print(f"✅ Found {data['count']} session(s)") + else: + print(f"❌ FAILED to list sessions - Status: {response.status_code}") + return False + + except Exception as e: + print(f"❌ FAILED to list sessions - Error: {e}") + return False + + # Stop monitoring + print("\n📍 Stopping monitoring session...") + try: + response = requests.post(f"{BASE_URL}/api/monitor/{session_id}/stop", timeout=5) + + if response.status_code == 200: + print(f"✅ Stopped monitoring session") + else: + print(f"❌ FAILED to stop monitoring - Status: {response.status_code}") + return False + + except Exception as e: + print(f"❌ FAILED to stop monitoring - Error: {e}") + return False + + return True + + +def main(): + """Run all tests""" + print("\n" + "="*60) + print("Multi-Token Price Monitor API - Test Suite") + print("="*60) + print(f"Testing API at: {BASE_URL}") + + results = [] + + # Run tests + results.append(("Health Check", test_health())) + if not results[-1][1]: + print("\n⚠️ Cannot continue - API is not running") + sys.exit(1) + + results.append(("Get Tokens", test_get_tokens())) + results.append(("Get Prices", test_get_price())) + results.append(("Check Threshold", test_check_threshold())) + results.append(("Monitoring Sessions", test_monitoring())) + + # Summary + print("\n" + "="*60) + print("Test Summary") + print("="*60) + + passed = sum(1 for _, result in results if result) + total = len(results) + + for test_name, result in results: + status = "✅ PASSED" if result else "❌ FAILED" + print(f"{status} - {test_name}") + + print(f"\nTotal: {passed}/{total} tests passed") + + if passed == total: + print("\n🎉 All tests passed! API is working correctly.") + print("\nNext steps:") + print("1. Read API_DOCUMENTATION.md for complete endpoint reference") + print("2. Deploy with ngrok: ngrok http 5000") + print("3. Start monitoring your favorite cryptocurrencies!") + else: + print("\n⚠️ Some tests failed. Please check the errors above.") + sys.exit(1) + + +if __name__ == "__main__": + if len(sys.argv) > 1: + BASE_URL = sys.argv[1] + + try: + main() + except KeyboardInterrupt: + print("\n\nTests interrupted by user") + sys.exit(1) +