Update FIGI parser

src/openfigi_parser.py

@@ -1,23 +1,35 @@
 # %%
+import os
 from dataclasses import dataclass
-from typing import Optional
+from typing import Literal, Union
+
 import pandas as pd
+
 import httpx
 
+from sqlalchemy import types, sql
+from azure import AzureDbConnection, ConnectionSettings
+
 
 # %%
 @dataclass
 class AssetInfo:
     FIGI: str
     Ticker: str
-    Title: str
-    Description: Optional[str]
-    AssetType: str = 'Cryptocurrency'
-    SourceId: str = "OpenFigi API"
-    Version: str = "v202204"
+    Title: Union[str, None]
+    Description: Union[str, None]
+    AssetType: Literal['Cryptocurrency']
+    SourceId: Literal['OpenFigi API']
+    Version: Literal['v202206']
 
 
-def get_asset_info(pair: str) -> AssetInfo:
+def get_figi(pair: str) -> Union[AssetInfo, None]:
+    """Return FIGI for pair
+
+    References:
+    - https://www.openfigi.com/assets/local/figi-allocation-rules.pdf
+    - https://www.openfigi.com/search
+    """
     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)
 
@@ -37,23 +49,23 @@ def get_asset_info(pair: str) -> AssetInfo:
     return pair_figi
 
 
-#%% Tests
+# %% Tests
 expected_pairs = {
     'WAX-USD': None,
     'ETH-USD': 'BBG00J3NBWD7',
     'BTC-USD': 'BBG006FCL7J4',
     'SOL-USD': 'BBG013WVY457',
-    'UNI-USD': 'BBG013TZFVW3'
+    'UNI-USD': 'BBG013TZFVW3',
+    'SUSHI-USD': 'KKG0000010W1'
 }
 
 for k, v in expected_pairs.items():
-    assert get_asset_info(k) == v
+    actual = get_figi(k)
+    print(actual)
+    assert actual == 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:
@@ -62,14 +74,44 @@ def insert_dash(text: str, position: int) -> str:
     else:
         return text
 
-usd_pairs = [insert_dash(s.upper(), 3) for s in pair_names if "usd" in s]
+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)
+pair_figi_list = [get_figi(p) for p in usd_pairs]
+
+# %% ----
+conn_settings = ConnectionSettings(server='***.database.windows.net', database='market-data-db', username='***', password='***')
+db_conn = AzureDbConnection(conn_settings)
+
+db_conn.connect()
+for t in db_conn.get_tables():
+    print(t)
+
+
 # %%
+db_mapping = {
+    'FIGI': types.VARCHAR(length=12),
+    'open': types.DECIMAL(precision=19, scale=9),
+    'high': types.DECIMAL(precision=19, scale=9),
+    'close': types.DECIMAL(precision=19, scale=9),
+    'low': types.DECIMAL(precision=19, scale=9),
+    'volume': types.DECIMAL(precision=19, scale=9),
+    'time': types.DATETIME(),
+    'source_id': types.SMALLINT,
+    'version': types.VARCHAR(length=12),
+    'interval': types.CHAR(length=2)
+}
+
+
+query = sql.text("select * from Cryptocurrency where figi = 'ustusd'")
+result = db_conn._conn.execute(query).fetchall()
+
+
+# %%
+db_conn.dispose()
+print('Completed')