Compare commits

...

6 Commits

Author SHA1 Message Date
codez0mb1e
b51d4a0162 Minor updates 2022-06-19 23:01:08 +00:00
codez0mb1e
153fc5a230 Parser update 2022-06-19 22:59:58 +00:00
codez0mb1e
a3a48fbcba OpenFigi crawler draft 2022-06-19 22:58:56 +00:00
codez0mb1e
0d5e961123 Update FIGI parser 2022-06-17 17:47:32 +00:00
codez0mb1e
e1f86b7714 Add .zip to except list 2022-06-17 16:51:48 +00:00
codez0mb1e
03dc615a8c Format connection string 2022-06-07 14:31:52 +00:00
5 changed files with 202 additions and 102 deletions

3
.gitignore vendored
View File

@ -1 +1,2 @@
/data/*.csv
/data/*.csv
/data/*.zip

View File

@ -1,16 +1,18 @@
# %% Import dependencies ----
from dataclasses import dataclass
from typing import Dict, Any, Iterable
from pandas import DataFrame
from typing import Dict, Any
from sqlalchemy import create_engine, inspect
import pandas as pd
import urllib
# %%
# %% Models
@dataclass(frozen=True)
class ConnectionSettings:
"""Connection Settings."""
"""Connection Settings"""
server: str
database: str
username: str
@ -19,10 +21,10 @@ class ConnectionSettings:
timeout: int = 30
# %% Connection
class AzureDbConnection:
"""
Azure SQL database connection.
"""
"""Azure SQL database connection."""
def __init__(self, conn_settings: ConnectionSettings, echo: bool = False) -> None:
conn_params = urllib.parse.quote_plus(
'Driver=%s;' % conn_settings.driver +
@ -36,29 +38,29 @@ class AzureDbConnection:
)
conn_string = f'mssql+pyodbc:///?odbc_connect={conn_params}'
self.db = create_engine(conn_string, echo=echo)
self._db = create_engine(conn_string, echo=echo)
def connect(self) -> None:
"""Estimate connection."""
self.conn = self.db.connect()
"""Estimate connection"""
self._conn = self._db.connect()
def get_tables(self) -> Iterable[str]:
"""Get list of tables."""
inspector = inspect(self.db)
def get_tables(self) -> list[str]:
"""Get list of tables"""
inspector = inspect(self._db)
return [t for t in inspector.get_table_names()]
def insert(self, inserted_data: DataFrame, target_table: str, db_mapping: Dict[str, Any], chunksize: int = 10000) -> None:
def insert(self, inserted_data: pd.DataFrame, target_table: str, db_mapping: Dict[str, Any], chunksize: int = 10000) -> None:
inserted_data.to_sql(
con=self.db,
con=self._db,
schema='dbo',
name=target_table,
if_exists='append', # or replace
if_exists='replace', # or append
index=False,
chunksize=chunksize,
dtype=db_mapping
)
def dispose(self) -> None:
"""Dispose opened connections."""
self.conn.close()
self.db.dispose()
"""Dispose opened connections"""
self._conn.close()
self._db.dispose()

View File

@ -1,7 +1,6 @@
#!/usr/bin/python3
"""
Data source: https://www.kaggle.com/code/tencars/bitfinexdataset
"""
@ -15,6 +14,11 @@ from azure import AzureDbConnection, ConnectionSettings
# %%
#> ~/apps/resistance/data
#> kaggle datasets download tencars/392-crypto-currency-pairs-at-minute-resolution
#> unzip 392-crypto-currency-pairs-at-minute-resolution.zip
input_path = "../data"
# Get names and number of available currency pairs
@ -28,6 +32,7 @@ print(pair_names[0:50])
usd_pairs = [s for s in pair_names if "usd" in s]
print(usd_pairs)
# %%
def load_data(symbol, source=input_path):
@ -46,12 +51,12 @@ def load_data(symbol, source=input_path):
# %% ----
solusd = load_data("solusd")
solusd.tail()
sample_df = load_data("ethusd")
sample_df
# %% ----
conn_settings = ...
db_conn = AzureDbConnection(conn_settings)
db_conn.connect()
@ -63,7 +68,7 @@ for t in db_conn.get_tables():
min_candels_n = 10000
db_mapping = {
'FIGI': types.VARCHAR(length=12),
'FIGI': types.CHAR(length=12),
'open': types.DECIMAL(precision=19, scale=9),
'high': types.DECIMAL(precision=19, scale=9),
'close': types.DECIMAL(precision=19, scale=9),
@ -82,14 +87,14 @@ for pair in usd_pairs:
candles_df['FIGI'] = pair
candles_df['time'] = candles_df.index
candles_df['source_id'] = 128
candles_df['version'] = 'v202204'
candles_df['version'] = 'v202206'
candles_df['interval'] = '1M'
if candles_df.shape[0] > min_candels_n:
print('{} rows from {} to {}'.format(candles_df.shape[0], min(candles_df['time']), max(candles_df['time'])))
print(f'Starting insert {pair}...')
db_conn.insert(candles_df, 'Cryptocurrency', db_mapping)
db_conn.insert(candles_df, 'crypto', db_mapping)
else:
print(f'WARN: {pair} has only {candles_df.shape[0]} records')

167
src/openfigi_crawler.py Normal file
View File

@ -0,0 +1,167 @@
# %% Import dependencies
import os
from dataclasses import dataclass
from typing import Dict, Union
import pandas as pd
import httpx
from sqlalchemy import types
from azure import AzureDbConnection, ConnectionSettings
# %% Data models
@dataclass
class AssetInfo:
FIGI: str
Ticker: str
Title: Union[str, None]
Description: Union[str, None]
AssetType: str = 'Cryptocurrency'
SourceId: str = 'OpenFigi API'
Version: str = 'v202206'
def as_dict(self) -> Dict[str, str]:
return {'Figi': self.FIGI, 'Ticker': self.Ticker}
# %% FIGI provider
class OpenFigiProvider:
"""
OpenFigi API provider
References:
https://www.openfigi.com/assets/local/figi-allocation-rules.pdf
https://www.openfigi.com/search
"""
@staticmethod
def _send_request(ticker: str, asset_type: str) -> pd.DataFrame:
api_url = f'https://www.openfigi.com/search/query?facetQuery=MARKET_SECTOR_DES:%22{asset_type}%22&num_rows=100&simpleSearchString={ticker}&start=0'
response = httpx.get(api_url)
json_response = response.json()
return pd.DataFrame.from_dict(json_response['result'], orient='columns')
@staticmethod
def _find_figi(df: pd.DataFrame, field_name: str) -> Union[str, None]:
if len(df) == 0 or field_name not in df.columns:
return None
result = df[field_name].dropna().unique()
if (len(result) != 1):
print(f'[WARN] Multiple ({len(result)}) FIGI records was found')
return None
return result[0]
@staticmethod
def _find_name(df: pd.DataFrame) -> Union[str, None]:
if len(df) == 0 or 'DS002_sd' not in df.columns:
return None
result = df['DS002_sd'].dropna().unique()
if (len(result) != 1):
print(f'[WARN] Multiple ({len(result)}) name records was found')
return None
return result[0]
def search(self, ticker: str, asset_type: str = 'Curncy') -> Union[AssetInfo, None]:
"""Return FIGI for pair"""
response_df = OpenFigiProvider._send_request(ticker, asset_type)
figi = OpenFigiProvider._find_figi(response_df, 'kkg_pairFIGI_sd')
if figi is None:
base_quote = ticker.split('-')[0]
print(f'[INFO] {ticker} > Try to search using base quote {base_quote}')
response_df = OpenFigiProvider._send_request(base_quote, asset_type)
figi = OpenFigiProvider._find_figi(response_df, 'kkg_baseAssetFigi_sd')
if figi is None:
return None
return AssetInfo(figi, ticker, None, None)
#%%
figi_provider = OpenFigiProvider()
assert figi_provider.search('WAX-USD') == None
assert figi_provider.search('ABCD') == None
# %% Tests
expected_pairs = {
'BNB-USD': 'KKG000007HZ5',
'ETH-USD': 'BBG00J3NBWD7',
'BTC-USD': 'BBG006FCL7J4',
'SOL-USD': 'BBG013WVY457',
'UNI-USD': 'BBG013TZFVW3',
'SUSHI-USD': 'KKG0000010W1',
'AVAX-USD': 'KKG000007J36'
}
for k, v in expected_pairs.items():
actual = figi_provider.search(k)
print(actual.as_dict())
assert (
isinstance(actual, AssetInfo)
and actual.FIGI == v
and actual.Ticker == k
)
# %% Get assets for searching figi
pair_names = [x[:-4] for x in os.listdir("../data")]
def insert_dash(text: str, position: int) -> str:
if '-' not in text:
return text[:position] + '-' + text[position:]
else:
return text
usd_pairs = [
insert_dash(s.upper(), 3)
for s in pair_names if "usd" in s
]
print(usd_pairs[1:10])
# %%
figi_provider = OpenFigiProvider()
pair_figi_list = [figi_provider.search(p) for p in usd_pairs]
# %% ----
db_conn = AzureDbConnection(conn_settings)
db_conn.connect()
for t in db_conn.get_tables():
print(t)
# %%
db_mapping = {
'Figi': types.CHAR(length=12),
'Ticker': types.VARCHAR(length=12)
}
figi_df = pd.DataFrame([t.as_dict() for t in pair_figi_list if isinstance(t, AssetInfo)])
db_conn.insert(figi_df, 'figi', db_mapping)
# %%
db_conn.dispose()
print('Completed')

View File

@ -1,75 +0,0 @@
# %%
from dataclasses import dataclass
from typing import Optional
import pandas as pd
import httpx
# %%
@dataclass
class AssetInfo:
FIGI: str
Ticker: str
Title: str
Description: Optional[str]
AssetType: str = 'Cryptocurrency'
SourceId: str = "OpenFigi API"
Version: str = "v202204"
def get_asset_info(pair: str) -> AssetInfo:
api_url = f'https://www.openfigi.com/search/query?facetQuery=MARKET_SECTOR_DES:%22Curncy%22&num_rows=100&simpleSearchString={pair}&start=0'
response = httpx.get(api_url)
json_response = response.json()
response_df = pd.DataFrame.from_dict(json_response['result'], orient='columns')
if len(response_df) == 0:
print(f'[WARN] {pair} not found')
return None
pair_figi = response_df.kkg_pairFIGI_sd.unique()
if (len(pair_figi) != 1):
print(f'[WARN] {len(pair_figi)} records was found for {pair}')
else:
print(f'[INFO] {pair} associated w/ FIGI {pair_figi[0]}')
return pair_figi
#%% Tests
expected_pairs = {
'WAX-USD': None,
'ETH-USD': 'BBG00J3NBWD7',
'BTC-USD': 'BBG006FCL7J4',
'SOL-USD': 'BBG013WVY457',
'UNI-USD': 'BBG013TZFVW3'
}
for k, v in expected_pairs.items():
assert get_asset_info(k) == v
# %%
import os
import pandas as pd
pair_names = [x[:-4] for x in os.listdir("../data")]
def insert_dash(text: str, position: int) -> str:
if '-' not in text:
return text[:position] + '-' + text[position:]
else:
return text
usd_pairs = [insert_dash(s.upper(), 3) for s in pair_names if "usd" in s]
print(usd_pairs)
# %%
pair_figi_list = [get_asset_info(p) for p in usd_pairs]
for p in usd_pairs:
print(p)
get_asset_info(p)
# %%