mirror of
https://github.com/wassname/options_backtester.git
synced 2026-06-28 03:34:12 +08:00
Catalina Syddall
73 lines
2.2 KiB
Python
73 lines
2.2 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:
|
||
msg_success = "👍 Successfully {}d {}".format(op, _symbol_str(done))
|
||
else:
|
||
msg_success = "No symbols where successfully scraped 😭"
|
||
if len(failed) > 0:
|
||
msg_fail = "⚠️️ Failed to {} {}: {}".format(op,
|
||
_symbol_str(len(failed)),
|
||
", ".join(failed))
|
||
else:
|
||
msg_fail = "No symbols failed to scrape! 🤩"
|
||
|
||
msg = msg_success + '\n' + msg_fail
|
||
slack_notification(msg, scraper, status=Status.Warning)
|
||
|
||
|
||
def _symbol_str(count):
|
||
return str(count) + " symbol" if count == 1 else str(count) + " symbols"
|