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
  20 import hjson
  21 import requests
  22 from requests.exceptions import ConnectionError, Timeout, TooManyRedirects
  23
  24 logging.basicConfig(format="%(asctime)s - %(levelname)s - %(message)s")
  25 BASE_URL = "https://pro-api.coinmarketcap.com/v1/{}"
  26
  27 """
  28 Every element must be the same structure:
  29
  30     sign_of_when_cell: (
  31         "the string used in \"currency costs {} it was expected\"",
  32         lambda asked, from_cap: whether_approaches(asked, from_cap)),
  33     )
  34
  35 """
  36 WHEN_TRANSLATER = {
  37     ">": ("more than", lambda asked, from_cap: from_cap > asked),
  38     "<": ("less than", lambda asked, from_cap: from_cap < asked),
  39 }
  40
  41
  42 def currencies(key: str):
  43     headers = {"Accepts": "application/json", "X-CMC_PRO_API_KEY": key}
  44     session = requests.Session()
  45     session.headers.update(headers)
  46
  47     url = BASE_URL.format("cryptocurrency/listings/latest?limit=5000")
  48     try:
  49         response = session.get(url)
  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):
  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         f = lambda from_cap: then_what if then_how(than_what, from_cap) else None
  72         return f
  73     except (IndexError, ValueError, KeyError):
  74         logging.error('Asked "when" is a trash: "%s".', s)
  75         return lambda x: None
  76
  77
  78 def main():
  79     try:
  80         document = hjson.load(sys.stdin)
  81         document["key"]
  82         document["watch"]
  83     except hjson.scanner.HjsonDecodeError as e:
  84         logging.error("Invalid syntax in stdin: %s", e)
  85         return 1
  86     except KeyError:
  87         logging.error('No "key" or "watch" cell in stdin.')
  88         return 1
  89
  90     listings = currencies(document["key"])
  91     if not listings:
  92         return 2
  93
  94     watching = document["watch"]
  95     functions = [None] * len(watching)
  96     for i, w in enumerate(watching):
  97         functions[i] = cmp_function(w["when"])
  98
  99     output = set()
 100     for currency in listings:
 101         for i in range(len(watching)):
 102             if watching[i]["symbol"] != currency["symbol"]:
 103                 continue
 104             try:
 105                 s = functions[i](currency["quote"]["USD"]["price"])
 106             except ValueError:
 107                 logging.error(
 108                     "Fetched currency listings are ugly. Where is the price placed?\n%s",
 109                     hjson.dumps(currency),
 110                 )
 111                 continue
 112             if s:
 113                 to_echo = s.format(
 114                     symbol=currency["symbol"], price=currency["quote"]["USD"]["price"]
 115                 )
 116                 output.add(to_echo)
 117
 118     for line in output:
 119         print(line)
 120
 121
 122 if __name__ == "__main__":
 123     status = main()
 124     if status:
 125         logging.info("Terminating...")
 126     sys.exit(status)