Minor updates

.gitignore

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

src/azure.py

@@ -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()

src/bitfinex_crypto_parser.py

@@ -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')
 

src/openfigi_crawler.py

@@ -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')

src/openfigi_parser.py

@@ -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)
-# %%