mirror of
https://github.com/wassname/options_backtester.git
synced 2026-07-07 10:55:55 +08:00
Catalina Syddall
73 lines
2.1 KiB
Python
73 lines
2.1 KiB
Python
import logging
|
|
from datetime import datetime
|
|
from enum import Enum
|
|
|
|
import requests
|
|
|
|
from .utils import get_module_config, get_environment_var
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
Status = Enum("Status", "Success Warning Error")
|
|
|
|
options = get_module_config("notifications")
|
|
try:
|
|
webhook = get_environment_var("SLACK_WEBHOOK")
|
|
except EnvironmentError as e:
|
|
logger.error("Missing slack webhook from configuration file")
|
|
raise e
|
|
|
|
payload = {
|
|
"channel": "#algotrading",
|
|
"username": "Talebot",
|
|
"icon_emoji": ":taleb:",
|
|
"attachments": [{
|
|
"footer": "Talebot"
|
|
}]
|
|
}
|
|
|
|
|
|
def slack_notification(text, scraper, status=Status.Error):
|
|
"""Post Slack notification"""
|
|
title = "data_scraper status report"
|
|
color="#ff9906"
|
|
msg = text
|
|
|
|
payload["attachments"][0]["fallback"] = msg
|
|
payload["attachments"][0]["text"] = msg
|
|
payload["attachments"][0]["color"] = color
|
|
payload["attachments"][0]["title"] = title
|
|
payload["attachments"][0]["fields"] = [{"title": scraper}]
|
|
payload["attachments"][0]["ts"] = datetime.today().timestamp()
|
|
|
|
response = requests.post(webhook, json=payload)
|
|
|
|
if response.status_code != 200:
|
|
msg = "Error connecting to Slack {}. Response is:\n{}".format(
|
|
response.status_code, response.text)
|
|
logger.error(msg)
|
|
|
|
|
|
def send_report(done, failed, scraper, op="scrape"):
|
|
"""Sends status report to Slack for given operation.
|
|
`done` is the count of successfully scraped/aggregated symbols
|
|
`failed` is a list of symbol names that could not be scraped/aggregated
|
|
"""
|
|
if done > 0:
|
|
msg1 = "Successfully {}d {}".format(op, _symbol_str(done))
|
|
if len(failed) > 0:
|
|
msg2 = "Failed to {} {}: {}".format(op, _symbol_str(len(failed)),
|
|
", ".join(failed))
|
|
|
|
msg= msg1 + " and " + msg2
|
|
slack_notification(msg, scraper, status=Status.Warning)
|
|
|
|
|
|
def _symbol_str(count):
|
|
return str(count) + " symbol" if count == 1 else str(count) + " symbols"
|
|
|
|
## make init + make env ##
|
|
##cambiar el channel
|
|
##sacar notificaciones por cada error y armar un mensaje combinando todo
|
|
##python
|
|
## |