Merge pull request #4 from hanchiang/feature/trading-view-webhook-tel…

…egram

Format vix central message and trading view message before sending to telegram

README.md

@@ -0,0 +1,9 @@
+# Introduction
+This repository sends market data notification to channels like telegram. Data sources are trading view and vix central.
+
+# Sample
+## Test message
+![test message](images/telegram_test_message.png)
+
+## Simulate real message
+![test message](images/telegram_simulate_real_traffic.png)

pyproject.toml

@@ -18,6 +18,7 @@ python-telegram-bot = {version = "^20.0b0", allow-prereleases = true}
 pytz = "^2022.7"
 pytest = "^7.2.0"
 pytest-asyncio = "^0.20.3"
+python-dotenv = "^0.21.0"
 
 [tool.poetry.dev-dependencies]
 debugpy = "^1.6.2"

src/config/config.py

@@ -0,0 +1,45 @@
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+def get_telegram_bot_token():
+    if not os.getenv('TELEGRAM_BOT_TOKEN', None):
+        raise RuntimeError("telegram bot token is missing")
+    return os.getenv('TELEGRAM_BOT_TOKEN')
+
+def get_telegram_channel_id():
+    if not os.getenv('TELEGRAM_CHANNEL_ID', None):
+        raise RuntimeError("telegram channel is missing")
+    return os.getenv('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_trading_view_ips():
+    if not os.getenv('TRADING_VIEW_IPS', None):
+        raise RuntimeError("trading view ips is missing")
+    return os.getenv('TRADING_VIEW_IPS').split(',')
+
+def get_telegram_admin_id():
+    if not os.getenv('TELEGRAM_ADMIN_ID', None):
+        raise RuntimeError("telegram admin id is missing")
+    return os.getenv('TELEGRAM_ADMIN_ID')
+
+def get_contango_single_day_decrease_alert_ratio():
+    return 0.4 if not get_is_testing_telegram() else 0.01
+
+def get_contango_decrease_past_n_days():
+    return 5 if not get_is_testing_telegram() else 2
+
+def get_potential_overextended_by_symbol():
+    potential_overextended_by_symbol = {
+        'SPY': {
+            'up': 0.04 if not get_is_testing_telegram() else 0.01,
+            'down': -0.065 if not get_is_testing_telegram() else -0.01
+        }
+    }
+    return potential_overextended_by_symbol

src/notification_destination/telegram_notification.py

@@ -0,0 +1,12 @@
+import telegram
+import src.config.config as config
+
+bot = telegram.Bot(token=config.get_telegram_bot_token())
+
+async def send_message_to_channel(message: str, chat_id = config.get_telegram_channel_id()):
+    res = await bot.send_message(chat_id, text=message, parse_mode='MarkdownV2')
+    return res
+
+async def send_message_to_admin(message: str, chat_id = config.get_telegram_admin_id()):
+    res = await bot.send_message(chat_id, text=message, parse_mode='MarkdownV2')
+    return res

src/server.py

@@ -1,9 +1,15 @@
-from fastapi import FastAPI
+from functools import reduce
+from typing import List, Any
+
+from fastapi import FastAPI, Request
 import uvicorn
 import os
 from src.dependencies import Dependencies
 from src.router.vix_central import thirdparty_vix_central, vix_central
 import src.notification_destination.telegram_notification as telegram_notification
+from src.service.vix_central import RecentVixFuturesValues
+import src.config.config as config
+from src.util.my_telegram import escape_markdown
 
 app = FastAPI()
 app.include_router(thirdparty_vix_central.router)
@@ -21,7 +27,7 @@ def start_server():
 
 @app.on_event("startup")
 async def startup_event():
-  Dependencies.build()
+  await Dependencies.build()
 @app.on_event("shutdown")
 async def shutdown_event():
     await Dependencies.cleanup()
@@ -30,7 +36,70 @@ async def shutdown_event():
 async def heath_check():
   return {"data": "Market data notification is running!"}
 
-# @app.post("/telegram")
-# async def heath_check():
-#   res = await telegram_notification.send_message("hello world")
-#   return {"data": f"Sent to {res.chat.title} {res.chat.type} at {res.date}"}
+@app.post("/tradingview-webhook")
+async def tradingview_webhook(request: Request):
+  # symbol, timeframe(e.g. 1d), close, ema20
+
+  print(f"{request.method} {request.url} Received request from {request.client}")
+
+  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"}
+
+  vix_central_service = Dependencies.get_vix_central_service()
+  vix_central_data = await vix_central_service.get_recent_values()
+  vix_central_message = format_vix_central_message(vix_central_data)
+
+  body = await request.json()
+  tradingview_message = format_tradingview_message(body)
+  tradingview_message = f"*Trading view market data:*{tradingview_message}"
+
+  messages = [vix_central_message, tradingview_message]
+  if config.get_is_testing_telegram():
+    messages.insert(0, '*THIS IS A TEST MESSAGE*')
+
+  telegram_message = escape_markdown("\n----------------------------\n").join(messages)
+
+  res = await telegram_notification.send_message_to_channel(message=telegram_message)
+  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):
+   message = reduce(format_vix_futures_values, vix_central_value.vix_futures_values,
+                               f"*VIX central data for {vix_central_value.vix_futures_values[0].futures_date} futures:*")
+   if vix_central_value.is_contango_decrease_for_past_n_days:
+     message = f"{message}\n*Contango has been decreasing for the past {vix_central_value.contango_decrease_past_n_days} days ‼️*"
+   return message
+
+
+def format_vix_futures_values(res, curr):
+  message = f"{res}\ndate: {escape_markdown(curr.current_date)}, contango %: {escape_markdown(curr.formatted_contango)}"
+  if curr.is_contango_single_day_decrease_alert:
+    threshold = f"{curr.contango_single_day_decrease_alert_ratio:.1%}"
+    message = f"{message}{escape_markdown('.')} *Contango changed by more than {escape_markdown(threshold)} from the previous day* ‼️"
+  return message
+
+def format_tradingview_message(payload: List[Any]):
+  message = ''
+  potential_overextended_by_symbol = config.get_potential_overextended_by_symbol()
+
+  for p in payload:
+    symbol = p['symbol'].upper()
+    close = p['close']
+    ema20 = p['ema20']
+    delta = (close - ema20) / ema20 if close > ema20 else -(ema20 - close) / ema20
+
+    delta_percent = f"{delta:.2%}"
+    message = f"{message}\nsymbol: {symbol}, close: {escape_markdown(str(close))}, ema20 1D: {escape_markdown(str(ema20))}, % change: {escape_markdown(delta_percent)}"
+
+    direction = 'up' if close > ema20 else 'down'
+    if potential_overextended_by_symbol.get(symbol, None) is not None:
+      if potential_overextended_by_symbol[symbol].get(direction) is not None:
+        overextended_threshold = potential_overextended_by_symbol[symbol][direction]
+        overextended_threshold_percent = f"{overextended_threshold:.2%}"
+        if abs(delta) > abs(overextended_threshold):
+          message = f"{message}, *greater than {escape_markdown(overextended_threshold_percent)} when it is {'above' if direction == 'up' else 'below'} the ema20, watch for potential rebound* ‼️"
+
+    return message

src/service/vix_central.py

@@ -1,12 +1,13 @@
 import datetime
 import asyncio
 from typing import List
+from src.config import config
 from src.third_party_service.vix_central import ThirdPartyVixCentralService
 import src.util.date_util as date_util
 
 
 class VixFuturesValue:
-    def __init__(self, contango_single_day_decrease_alert_ratio = 0.4):
+    def __init__(self, contango_single_day_decrease_alert_ratio):
         # "yyyy-mm-dd". 2022-12-30
         self.current_date: str = None
         # "yyyy mmm". e.g. 2022 Jan
@@ -23,7 +24,7 @@ def __init__(self, contango_single_day_decrease_alert_ratio = 0.4):
         self.contango_single_day_decrease_alert_ratio: float = contango_single_day_decrease_alert_ratio
 
 class RecentVixFuturesValues:
-    def __init__(self, decrease_past_n_days = 5):
+    def __init__(self, decrease_past_n_days):
         # store values of the upcoming VIX futures(next month)
         # list of dict in reverse chronological order of current_date.
         self.vix_futures_values: List[VixFuturesValue] = []
@@ -35,8 +36,8 @@ class VixCentralService:
     VALUE_CAPACITY = 5
     # month of the vix futures we are interested in. e.g. "Jan"
     MONTH_OF_INTEREST = None
-    CONTANGO_SINGLE_DAY_DECREASE_ALERT_RATIO = 0.01
-    CONTANGO_DECREASE_PAST_N_DAYS = 1
+    CONTANGO_SINGLE_DAY_DECREASE_ALERT_RATIO = config.get_contango_single_day_decrease_alert_ratio()
+    CONTANGO_DECREASE_PAST_N_DAYS = config.get_contango_decrease_past_n_days()
 
     def __init__(self, third_party_service = ThirdPartyVixCentralService):
         self.third_party_service = third_party_service
@@ -106,8 +107,10 @@ def compute_contango_alert_threshold(self, recent_values: RecentVixFuturesValues
                 recent_values.vix_futures_values[i].is_contango_single_day_decrease_alert = False
 
             if decrease_counter < recent_values.contango_decrease_past_n_days and curr_contango < prev_contango:
-                is_decrease_for_past_n_days = True
-            decrease_counter += 1
+                decrease_counter += 1
+                if decrease_counter == recent_values.contango_decrease_past_n_days:
+                    is_decrease_for_past_n_days = True
+
         # No previous item to compare to, so it is always false
         recent_values.vix_futures_values[len(recent_values.vix_futures_values) - 1].is_contango_single_day_decrease_alert = False
 

src/util/my_telegram.py

@@ -0,0 +1,4 @@
+import telegram
+
+def escape_markdown(text: str, version=2):
+    return telegram.helpers.escape_markdown(text=text, version=version)