nuke_alert.py

"""
Extremely hacky code, but it should work.
"""
import itertools
import random
import json
import os
import time
import urllib.parse
import urllib.request

# pip install vonage
import vonage

"""
make a credentials.py with the following variables:
VONAGE_API_KEY = ""
VONAGE_API_SECRET = ""
VONAGE_APPLICATION_ID = ""
VONAGE_PRIVATE_KEY_PATH = ""
MY_PHONE_NUMBER = ""
VONAGE_PHONE_NUMBER = ""
TELEGRAM_TOKEN = ""
TELEGRAM_CHAT_ID = ""
PHONE_NUMBERS = ["", ""]
"""

import credentials


# Call everyone if max probability is >= this
P_THRESHOLD = 0.3



def call_numbers():
    message = (
        "This is an alert. A prediction market, on nuclear weapon use, passed the 30% threshold."
        " Please keep calm. And double check that this is not a false alarm. Have a great day!"
    )
    client = vonage.Client(
        key=credentials.VONAGE_API_KEY,
        secret=credentials.VONAGE_API_SECRET,
        application_id=credentials.VONAGE_APPLICATION_ID,
        private_key=credentials.VONAGE_PRIVATE_KEY_PATH,
    )

    voice = vonage.Voice(client)

    # Call me first and give me 2 mins to check if it's a false alarm
    voice.create_call(
        {
            "from": {"type": "phone", "number": credentials.VONAGE_PHONE_NUMBER},
            "to": [{"type": "phone", "number": credentials.MY_PHONE_NUMBER}],
            "ncco": [{"action": "talk", "text": message}],
        }
    )

    time.sleep(120)

    numbers = [
        {"type": "phone", "number": number} for number in credentials.PHONE_NUMBERS
    ]

    for number in numbers:
        # See https://api.support.vonage.com/hc/en-us/articles/360015794051
        # See https://developer.vonage.com/en/voice/voice-api/code-snippets/making-calls/make-an-outbound-call-with-ncco?source=voice&lang=python
        # See https://api.support.vonage.com/hc/en-us/articles/207100288-How-many-calls-can-I-create-per-second-
        voice.create_call(
            {
                "from": {"type": "phone", "number": credentials.VONAGE_PHONE_NUMBER},
                "to": [number],
                "ncco": [{"action": "talk", "text": message}],
            }
        )
        time.sleep(0.5)
    # Super hacky: if you reach this, don't run again
    os.rename(__file__, __file__ + "_run")


# Send me a bunch of telegram messages before calling
def telegram_message(msg):
    msg = urllib.parse.quote(msg)
    url = f"https://api.telegram.org/bot{credentials.TELEGRAM_TOKEN}/sendMessage?chat_id={credentials.TELEGRAM_CHAT_ID}&text={msg}"
    for _ in range(4):
        try:
            urllib.request.urlopen(url)
        except Exception as e:
            print(f"NUKE ALERT TELEGRAM EXCEPTION: {e!r}")

        time.sleep(5)


def nuke_alert():
    telegram_message("NUKE ALERT!")
    time.sleep(120)
    call_numbers()


def get_json(url, retries=3):
    try:
        response = urllib.request.urlopen(url)
        data = response.read()
        return json.loads(data)
    except Exception as e:
        if retries > 0:
            time.sleep(7 + random.random() * 37)
            return get_json(url, retries - 1)
        else:
            telegram_message(f"NUKE ALERT EXCEPTION: {e!r} for {url}")
            return None


def get_manifold_data():
    urls = [
        "https://manifold.markets/jack/will-a-nuclear-weapon-detonate-in-n-6edbcd23a9f1",
        "https://manifold.markets/AndyMartin/will-a-nuclear-weapon-be-launched-i-015e44ed91f5",
        "https://manifold.markets/AmmonLam/will-a-nuclear-weapon-cause-over-10",
    ]
    for url in urls:
        slug = url.split("/")[-1]
        data = get_json(f"https://manifold.markets/api/v0/slug/{slug}")
        if data is not None:
            yield data.get("probability", 0)


def get_metaculus_data():
    q_ids = [13933, 7404, 2797, 13171]
    for q_id in q_ids:
        data = get_json(f"https://www.metaculus.com/api2/questions/{q_id}/")
        if data is None:
            continue
        metaculus_pred = data.get("metaculus_prediction", {}).get("full", 0)
        community_pred = (
            data.get("community_prediction", {}).get("full", {}).get("q2", 0)
        )
        if q_id == 2797:  # Question is negated
            metaculus_pred = 1 - metaculus_pred if metaculus_pred != 0 else 0
            community_pred = 1 - community_pred if community_pred != 0 else 0
        yield max(metaculus_pred, community_pred)


try:
    max_p = max(itertools.chain(get_metaculus_data(), get_manifold_data()))
    # print(f"Nuke max p: {max_p}")
    if max_p >= P_THRESHOLD:
        nuke_alert()
except Exception as e:
    print(f"NUKE ALERT EXCEPTION: {e!r}")
    telegram_message(f"NUKE ALERT EXCEPTION: {e!r}")