"""API for Numerai Signals"""
from typing import List, Dict, Tuple, Union
import os
import decimal
from io import BytesIO
import requests
import pandas as pd
from numerapi import base_api
from numerapi import utils
[docs]
class SignalsAPI(base_api.Api):
""""API for Numerai Signals"""
def __init__(self, *args, **kwargs):
base_api.Api.__init__(self, *args, **kwargs)
self.tournament_id = 11
[docs]
def get_leaderboard(self, limit: int = 50, offset: int = 0) -> List[Dict]:
"""Get the current Numerai Signals leaderboard
Args:
limit (int): number of items to return (optional, defaults to 50)
offset (int): number of items to skip (optional, defaults to 0)
Returns:
list of dicts: list of leaderboard entries
Each dict contains the following items:
* username (`str`)
* sharpe (`float`)
* rank (`int`)
* prevRank (`int`)
* today (`float`)
* mmc (`float`)
* mmcRank (`int`)
* icRep (`float`)
* icRank (`int`)
* tcRep (`float`)
* tcRank (`int`)
* nmrStaked (`float`)
Example:
>>> numerapi.SignalsAPI().get_leaderboard(1)
[{'prevRank': 1,
'rank': 1,
'sharpe': 2.3,
'today': 0.01321,
'username': 'floury_kerril_moodle',
'mmc': -0.0101202715,
'mmcRank': 30,
'nmrStaked': 13.0,
'icRep': -0.0101202715,
'icRank': 30,
..
}]
"""
query = '''
query($limit: Int!
$offset: Int!) {
signalsLeaderboard(limit: $limit
offset: $offset) {
prevRank
rank
sharpe
today
username
mmc
mmcRank
nmrStaked
icRank
icRep
tcRep
tcRank
}
}
'''
arguments = {'limit': limit, 'offset': offset}
data = self.raw_query(query, arguments)['data']['signalsLeaderboard']
return data
[docs]
def upload_predictions(self, file_path: str = "predictions.csv",
model_id: str = None,
df: pd.DataFrame = None,
timeout: Union[None, float, Tuple[float, float]] = (10, 60),
) -> str:
"""Upload predictions from file.
Will read TRIGGER_ID from the environment if this model is enabled with
a Numerai Compute cluster setup by Numerai CLI.
Args:
file_path (str): CSV file with predictions that will get uploaded
model_id (str): Target model UUID (required for accounts
with multiple models)
df (pandas.DataFrame): Pandas DataFrame to upload, if function is
given df and file_path, df will be uploaded
timeout (float|tuple(float,float)): waiting time (connection timeout,
read timeout)
Returns:
str: submission_id
Example:
>>> api = SignalsAPI(secret_key="..", public_id="..")
>>> model_id = api.get_models()['uuazed']
>>> api.upload_predictions("prediction.cvs", model_id=model_id)
'93c46857-fed9-4594-981e-82db2b358daf'
>>> # upload directly from a pandas DataFrame:
>>> api.upload_predictions(df = predictions_df, model_id=model_id)
"""
self.logger.info("uploading predictions...")
# write the pandas DataFrame as a binary buffer if provided
buffer_csv = None
if df is not None:
buffer_csv = BytesIO(df.to_csv(index=False).encode())
buffer_csv.name = file_path
auth_query = '''
query($filename: String!
$modelId: String) {
submissionUploadSignalsAuth(filename: $filename
modelId: $modelId) {
filename
url
}
}
'''
arguments = {'filename': os.path.basename(file_path),
'modelId': model_id}
submission_resp = self.raw_query(auth_query, arguments,
authorization=True)
auth = submission_resp['data']['submissionUploadSignalsAuth']
# get compute id if available and pass it along
headers = {"x_compute_id": os.getenv("NUMERAI_COMPUTE_ID")}
with open(file_path, 'rb') if df is None else buffer_csv as file:
requests.put(auth['url'], data=file.read(),
headers=headers, timeout=timeout)
create_query = '''
mutation($filename: String!
$modelId: String
$triggerId: String) {
createSignalsSubmission(filename: $filename
modelId: $modelId
triggerId: $triggerId
source: "numerapi") {
id
firstEffectiveDate
}
}
'''
arguments = {'filename': auth['filename'],
'modelId': model_id,
'triggerId': os.getenv('TRIGGER_ID', None)}
create = self.raw_query(create_query, arguments, authorization=True)
return create['data']['createSignalsSubmission']['id']
[docs]
def submission_status(self, model_id: str = None) -> None:
"""submission status of the last submission associated with the account
DEPRECATED numerai no longer provides this data. This will be removed
in one of the next versions
Args:
model_id (str)
Example:
>>> api = SignalsAPI(secret_key="..", public_id="..")
>>> model_id = api.get_models()['uuazed']
>>> api.submission_status(model_id)
"""
_ = model_id
self.logger.warning("Method submission_status is DEPRECATED and will be removed soon.")
[docs]
def public_user_profile(self, username: str) -> Dict:
"""Fetch the public Numerai Signals profile of a user.
Args:
username (str)
Returns:
dict: user profile including the following fields:
* username (`str`)
* startDate (`datetime`)
* id (`string`)
* bio (`str`)
* nmrStaked (`decimal.Decimal`)
Example:
>>> api = SignalsAPI()
>>> api.public_user_profile("floury_kerril_moodle")
{'bio': None,
'id': '635db2a4-bdc6-4e5d-b515-f5120392c8c9',
'startDate': datetime.datetime(2019, 3, 26, 0, 43),
'username': 'floury_kerril_moodle',
'nmrStaked': Decimal('14.630994874320760131')}
"""
query = """
query($username: String!) {
v2SignalsProfile(modelName: $username) {
id
startDate
username
bio
nmrStaked
}
}
"""
arguments = {'username': username}
data = self.raw_query(query, arguments)['data']['v2SignalsProfile']
# convert strings to python objects
utils.replace(data, "startDate", utils.parse_datetime_string)
utils.replace(data, "nmrStaked", utils.parse_float_string)
return data
[docs]
def ticker_universe(self) -> List[str]:
"""fetch universe of accepted tickers
Returns:
list of strings: list of currently accepted tickers
Example:
>>> SignalsAPI().ticker_universe()
["MSFT", "AMZN", "APPL", ...]
"""
path = self.download_dataset("signals/universe_public.csv")
return pd.read_csv(path).ticker.tolist()
[docs]
def download_validation_data(self) -> None:
"""download CSV file with historical targets and ticker universe
"""
self.logger.warning("'download_validation_data' is DEPRECATED.")
self.logger.warning("Please use 'download_dataset' and 'list_datasets'")
[docs]
def stake_get(self, username) -> decimal.Decimal:
"""get current stake for a given users
Args:
username (str)
Returns:
decimal.Decimal: current stake
Example:
>>> SignalsAPI().stake_get("uuazed")
Decimal('14.63')
"""
data = self.public_user_profile(username)
return data['totalStake']
