58 lines
2.1 KiB
Python
58 lines
2.1 KiB
Python
import logging
|
|
from dotenv import load_dotenv
|
|
import requests
|
|
import os
|
|
import json
|
|
from decimal import Decimal
|
|
|
|
load_dotenv()
|
|
|
|
|
|
class JSONEncoder_Decimal(json.JSONEncoder):
|
|
def default(self, o):
|
|
if isinstance(o, Decimal):
|
|
return float(o)
|
|
return super(JSONEncoder_Decimal, self).default(o)
|
|
|
|
def upsert_list_of_dicts_by_id(list_of_dicts, new_dict, id='id', seq_check_field: str | None = None, reset_seq_id: bool = False) -> list[dict]:
|
|
for index, item in enumerate(list_of_dicts):
|
|
if item.get(id) == new_dict.get(id):
|
|
if ( seq_check_field is not None ) and ( not(reset_seq_id) ):
|
|
if item.get(seq_check_field) > new_dict.get(seq_check_field):
|
|
logging.info('Skipping out of sequence msg')
|
|
return list_of_dicts
|
|
list_of_dicts[index] = new_dict
|
|
return list_of_dicts
|
|
|
|
list_of_dicts.append(new_dict)
|
|
return list_of_dicts
|
|
|
|
def send_tg_alert(msg: str):
|
|
token = os.getenv("TG_TOKEN")
|
|
chat_id = os.getenv("TG_ALERTS_CHAT_ID")
|
|
|
|
url = f'https://api.telegram.org/bot{token}/sendMessage?chat_id={chat_id}'
|
|
response = requests.post(url, json={'text': str(str(msg)[:250])}, timeout=10)
|
|
|
|
return response.json()
|
|
|
|
def rec_set_dict(orig_dict, new_dict, allow_new_fields: bool = False) -> dict:
|
|
for k, v in new_dict.items():
|
|
if isinstance(v, dict):
|
|
rec_set_dict(orig_dict=orig_dict[k], new_dict=v)
|
|
else:
|
|
if allow_new_fields:
|
|
orig_dict[k] = v
|
|
else:
|
|
if orig_dict.get(k, None) is not None:
|
|
orig_dict[k] = v
|
|
else:
|
|
logging.warning(msg=f'rec_set_dict: encountered nonexistent key: "{k}"; skipping')
|
|
|
|
return orig_dict
|
|
|
|
def symbol_to_aster_fmt(symbol: str) -> str:
|
|
return (symbol+'T' if symbol[-1].upper()!='T' else symbol).replace('-','').upper()
|
|
|
|
def symbol_to_extend_fmt(symbol: str) -> str:
|
|
return (symbol[0:-1] if symbol[-1].upper()=='T' else symbol).replace('-','').upper().split('USD')[0]+'-'+'USD' |