I can’t the python script to work for connecting tradingview webhooks to ironbeam for automated trading. Please help or advise. Thanks!
!pip install Flask pyngrok
from flask import Flask, request, jsonify
from pyngrok import ngrok
from datetime import datetime, timedelta
import requests
app = Flask(name)
Initialize last_webhook_time to a time in the past
last_webhook_time = datetime.now() - timedelta(days=1)
Set your ngrok authtoken
ngrok.set_auth_token(“####################################”) # Replace with your actual authtoken
Start Ngrok
public_url = ngrok.connect(5000)
print(“Ngrok tunnel "{}" → "http://127.0.0.1:5000"”.format(public_url))
Ironbeam API credentials
account_id = ‘##########’ # Replace with your actual account ID
password = ‘#######’ # Replace with your actual password
api_key = ‘###########################’ # Replace with your actual API key
Authenticate and get the token
def authenticate():
url = “https://demo.ironbeamapi.com/v2/auth”
payload = {
“username”: account_id,
“password”: password,
“apiKey”: api_key
}
headers = {“Content-Type”: “application/json”}
response = requests.post(url, json=payload, headers=headers)
data = response.json()
if response.status_code == 200:
print("Authentication successful.")
return data['token']
else:
print("Authentication failed:", response.status_code, data)
return None
Get the token for subsequent requests
token = authenticate()
@app.route(‘/’)
def home():
return “Hello, Ngrok!”
@app.route(‘/webhook’, methods=[‘POST’])
def webhook():
global last_webhook_time # Access the global variable
last_webhook_time = datetime.now() # Update with current time
data = request.json
print("Received data:", data) # Log the received data for debugging
symbol = data.get('symbol')
action = data.get('action')
quantity = data.get('quantity', 1) # Default to 1 if not provided
if not symbol or not action:
return jsonify({"status": "error", "message": "Missing symbol or action"}), 400
order_data = {
"symbol": symbol,
"action": action,
"quantity": quantity,
"orderType": "Market", # Change as needed (e.g., Limit, Stop)
"timeInForce": "GTC" # Good 'Til Canceled
}
headers = {
'Authorization': f'Bearer {token}', # Use the token for authorization
'Content-Type': 'application/json'
}
print("Sending order data:", order_data) # Log the order data being sent
try:
response = requests.post("https://demo.ironbeamapi.com/v2/orders", json=order_data, headers=headers)
# Log the response for debugging
print("Ironbeam response status code:", response.status_code)
print("Ironbeam response body:", response.text)
if response.status_code == 200:
print("Order executed successfully:", response.json())
return jsonify({"status": "success", "data": response.json()}), 200
else:
# Log the error details
error_message = response.json().get('message', 'No error message provided')
print("Error executing order:", response.status_code, error_message)
return jsonify({"status": "error", "message": error_message}), response.status_code
except requests.exceptions.RequestException as e:
# Handle any exceptions that occur during the request
print("Request to Ironbeam failed:", str(e))
return jsonify({"status": "error", "message": "Request to Ironbeam failed", "details": str(e)}), 500
def has_recent_webhook(time_window_seconds=60):
“”“Checks if a webhook was received within the given time window.”“”
time_difference = datetime.now() - last_webhook_time
return time_difference.total_seconds() <= time_window_seconds
Example usage:
if has_recent_webhook():
print(“A webhook signal was received within the last minute!”)
else:
print(“No recent webhook signals.”)
if name == ‘main’:
app.run(port=5000, use_reloader=False) # Keep the port consistent