Add disable telegram flag, and trading view webhook authentication

README.md

@@ -10,4 +10,4 @@ This repository sends market data notification to channels like telegram. Data s
 ![test message](images/telegram_test_message.png)
 
 ## Simulate real message
-![test message](images/telegram_simulate_real_traffic.png)
+![test message](images/telegram_simulate_tradingview_traffic.png)

src/config/config.py

@@ -21,8 +21,8 @@ def get_telegram_channel_id():
 def get_is_testing_telegram():
     return os.getenv('IS_TESTING_TELEGRAM', False) == 'true'
 
-def get_simulate_real_traffic():
-    return os.getenv('SIMULATE_REAL_TRAFFIC', False) == 'true'
+def get_simulate_tradingview_traffic():
+    return os.getenv('SIMULATE_TRADINGVIEW_TRAFFIC', False) == 'true'
 
 def get_trading_view_ips():
     if not os.getenv('TRADING_VIEW_IPS', None):
@@ -50,4 +50,12 @@ def get_potential_overextended_by_symbol():
             'down': -0.065 if not get_is_testing_telegram() else -0.01
         }
     }
-    return potential_overextended_by_symbol
+    return potential_overextended_by_symbol
+
+def get_tradingview_webhook_secret():
+    if not os.getenv('TRADING_VIEW_WEBHOOK_SECRET', None):
+        raise RuntimeError('TRADING_VIEW_WEBHOOK_SECRET is missing')
+    return os.getenv('TRADING_VIEW_WEBHOOK_SECRET')
+
+def get_disable_telegram():
+    return os.getenv('DISABLE_TELEGRAM', False) == 'true'

src/notification_destination/telegram_notification.py

@@ -5,6 +5,9 @@
 admin_bot = telegram.Bot(token=config.get_telegram_admin_bot_token())
 
 async def send_message_to_channel(message: str, chat_id = config.get_telegram_channel_id()):
+    if config.get_disable_telegram():
+        print('Telegram is disabled')
+        return
     res = await bot.send_message(chat_id, text=message, parse_mode='MarkdownV2')
     return res
 

src/server.py

@@ -1,3 +1,4 @@
+import asyncio
 from functools import reduce
 from typing import List, Any
 
@@ -38,16 +39,20 @@ async def heath_check():
 
 @app.post("/tradingview-webhook")
 async def tradingview_webhook(request: Request):
-  # symbol, timeframe(e.g. 1d), close, ema20
-
+    # Body is a list of: symbol, timeframe(e.g. 1d), close, ema20
   print(f"{request.method} {request.url} Received request from {request.client}")
 
+  if request.headers.get('x-tradingview-webhook-secret', None) != config.get_tradingview_webhook_secret():
+      message = f"[Potential malicious request warning]‼️ Trading view webhook secret {request.headers.get('x-tradingview-webhook-secret', None)} is incorrect"
+      asyncio.create_task(telegram_notification.send_message_to_admin(message=message))
+      return {"data": "OK"}
+
   trading_view_ips = config.get_trading_view_ips()
-  if not config.get_is_testing_telegram() and not config.get_simulate_real_traffic() and request.client.host not in trading_view_ips:
-    message = f"Warning‼️ Request ip {request.client.host} is not from trading view {trading_view_ips}"
-    await telegram_notification.send_message_to_admin(message=escape_markdown(message))
-    print(message)
-    return {"data": "OK"}
+  if not config.get_is_testing_telegram() and not config.get_simulate_tradingview_traffic() and request.client.host not in trading_view_ips:
+      message = f"[Potential malicious request warning]‼️Request ip {request.client.host} is not from trading view {trading_view_ips}"
+      print(message)
+      asyncio.create_task(telegram_notification.send_message_to_admin(message=escape_markdown(message)))
+      return {"data": "OK"}
 
   vix_central_service = Dependencies.get_vix_central_service()
   vix_central_data = await vix_central_service.get_recent_values()
@@ -64,6 +69,9 @@ async def tradingview_webhook(request: Request):
   telegram_message = escape_markdown("\n-----------------------------------------------------------------\n").join(messages)
 
   res = await telegram_notification.send_message_to_channel(message=telegram_message)
+  if not res:
+      return {"data": "OK"}
+  print(f"Sent to {res.chat.title} {res.chat.type} at {res.date}. Message id {res.id}")
   return {"data": f"Sent to {res.chat.title} {res.chat.type} at {res.date}. Message id {res.id}"}
 
 def format_vix_central_message(vix_central_value: RecentVixFuturesValues):