cryptoalert

   1 #!/usr/bin/env python
   2
   3 # Copyright 2021 Oskar Sharipov <oskarsh[at]riseup[dot]net
   4 #
   5 # Licensed under the Apache License, Version 2.0 (the "License");
   6 # you may not use this file except in compliance with the License.
   7 # You may obtain a copy of the License at
   8 #
   9 #   http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 # Unless required by applicable law or agreed to in writing, software
  12 # distributed under the License is distributed on an "AS IS" BASIS,
  13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16
  17 import logging
  18 import sys
  19 from typing import List, Optional
  20
  21 import hjson
  22 import requests
  23 from requests.exceptions import ConnectionError, Timeout, TooManyRedirects
  24
  25 logging.basicConfig(format="%(asctime)s - %(levelname)s - %(message)s")
  26 BASE_URL = "https://pro-api.coinmarketcap.com/v1/{}"
  27
  28 """
  29 Every element must be the same structure:
  30
  31     "sign_of_when_cell": (
  32         "the string used in \"currency costs {} it was expected\"",
  33         lambda asked, from_cap: whether_approaches(asked, from_cap)),
  34     )
  35
  36 """
  37 WHEN_TRANSLATER = {
  38     ">": ("more than", lambda asked, from_cap: from_cap > asked),
  39     "<": ("less than", lambda asked, from_cap: from_cap < asked),
  40 }
  41
  42
  43 def currencies(key: str, symbols: List[str]):
  44     headers = {"Accepts": "application/json", "X-CMC_PRO_API_KEY": key}
  45     params = {"symbol": ",".join(symbols)}
  46
  47     url = BASE_URL.format("cryptocurrency/quotes/latest")
  48     try:
  49         response = requests.get(url, params=params, headers=headers)
  50         j = response.json()
  51         if "data" not in j:
  52             logging.error(
  53                 'Couldn\'t fetch currency listings. Possibly "key" is invalid.'
  54             )
  55             return {}
  56         return j["data"]
  57     except (ConnectionError, Timeout, TooManyRedirects) as e:
  58         logging.error("Couldn't fetch currency listings: %s.", e)
  59         return {}
  60
  61
  62 def cmp_function(s: str, msg: Optional[str] = None):
  63     try:
  64         than_what = float(s[1:])
  65         then_what, then_how = WHEN_TRANSLATER[s[0]]
  66         then_what = (
  67             "{symbol} costs "
  68             + then_what
  69             + " it was expected. Its current price is {price:.4f} USD."
  70         )
  71         if msg:
  72             then_what += f" Message: {msg}"
  73         f = lambda from_cap: then_what if then_how(than_what, from_cap) else None
  74         return f
  75     except (IndexError, ValueError, KeyError):
  76         logging.error('Asked "when" is a trash: "%s".', s)
  77         return lambda x: None
  78
  79
  80 def main():
  81     try:
  82         document = hjson.load(sys.stdin)
  83         document["key"]
  84         document["watch"]
  85     except hjson.scanner.HjsonDecodeError as e:
  86         logging.error("Invalid syntax in stdin: %s", e)
  87         return 1
  88     except KeyError:
  89         logging.error('No "key" or "watch" cell in stdin.')
  90         return 1
  91
  92     asked = set()
  93     for el in document["watch"]:
  94         asked.add(el["symbol"])
  95
  96     quotes = currencies(document["key"], list(asked))
  97     if not quotes:
  98         return 2
  99
 100     watching = document["watch"]
 101     functions = [None] * len(watching)
 102     for i, w in enumerate(watching):
 103         functions[i] = cmp_function(w["when"], w.get("msg"))
 104
 105     output = set()
 106     for currency in quotes.values():
 107         for i in range(len(watching)):
 108             if watching[i]["symbol"] != currency["symbol"]:
 109                 continue
 110             try:
 111                 s = functions[i](currency["quote"]["USD"]["price"])
 112             except ValueError:
 113                 logging.error(
 114                     "Fetched currency listings are ugly. Where is the price placed?\n%s",
 115                     hjson.dumps(currency),
 116                 )
 117                 continue
 118             if s:
 119                 to_echo = s.format(
 120                     symbol=currency["symbol"], price=currency["quote"]["USD"]["price"]
 121                 )
 122                 output.add(to_echo)
 123
 124     for line in output:
 125         print(line)
 126
 127
 128 if __name__ == "__main__":
 129     status = main()
 130     if status:
 131         logging.info("Terminating...")
 132     sys.exit(status)