CachedTokenSniffer
Documentation for eth_defi.token_analysis.tokensniffer.CachedTokenSniffer Python class.
- class CachedTokenSniffer[source]
Add file-system based cache for TokenSniffer API.
See
TokenSniffer
class for detailsUse SQLite DB as a key-value cache backend, or your custom cache interface
No support for multithreading/etc. fancy stuff
Example usage:
from eth_defi.token_analysis.tokensniffer import CachedTokenSniffer, is_tradeable_token # # Setup TokenSniffer # db_file = Path(cache_path) / "tokensniffer.sqlite" tokensniffer_threshold = 24 # Quite low threshold, 0 = total scam sniffer = CachedTokenSniffer( db_file, TOKENSNIFFER_API_KEY, ) ticker = make_full_ticker(pair_metadata[pair_id]) address = pair_metadata[pair_id]["base_token_address"] sniffed_data = sniffer.fetch_token_info(chain_id.value, address) if not is_tradeable_token(sniffed_data, risk_score_threshold=tokensniffer_threshold): score = sniffed_data["score"] print(f"WARN: Skipping pair {ticker} as the TokenSniffer score {score} is below our risk threshold") continue
You can also use your own cache interface instead of SQLite. Here is an example SQLALchemy implementation:
class TokenInternalCache(UserDict): def __init__(self, dbsession: Session): self.dbsession = dbsession def match_token(self, token_spec: str) -> Token: # Sniffer interface gives us tokens as {chain}-{address} strings chain, address = token_spec.split("-") chain_id = int(chain) address = HexBytes(address) return self.dbsession.query(Token).filter(Token.chain_id == chain_id, Token.address == address).one_or_none() def __getitem__(self, name) -> None | str: token = self.match_token(name) if token is not None: if token.etherscan_data is not None: return token.etherscan_data.get("tokensniffer_data") return None def __setitem__(self, name, value): token = self.match_token(name) if token.etherscan_data is None: token.etherscan_data = {} token.etherscan_data["tokensniffer_data"] = value def __contains__(self, key): return self.get(key) is not None # And then usage: weth = dbsession.query(Token).filter_by(symbol="WETH", chain_id=1).one() sniffer = CachedTokenSniffer( cache_file=None, api_key=TOKENSNIFFER_API_KEY, cache=cast(dict, TokenInternalCache(dbsession)), ) data = sniffer.fetch_token_info(weth.chain_id, weth.address.hex()) assert data["cached"] is False data = sniffer.fetch_token_info(weth.chain_id, weth.address.hex()) assert data["cached"] is True
Methods summary
__init__
(cache_file, api_key[, session, cache])- param api_key
fetch_token_info
(chain_id, address)Get TokenSniffer info.
Get a diagnostics message.
- __init__(cache_file, api_key, session=None, cache=None)[source]
- Parameters
api_key (str) – TokenSniffer API key.
session (Optional[requests.sessions.Session]) – requests.Session for persistent HTTP connections
cache_file (pathlib.Path | None) –
Path to a local file system SQLite file used as a cached.
For simple local use cases.
Direct custom cache interface as a Python dict interface.
For your own database caching.
Cache keys are format: cache_key = f”{chain_id}-{address}”. Cache values are JSON blobs as string.
- fetch_token_info(chain_id, address)[source]
Get TokenSniffer info.
Use local file cache if available.
- Returns
Data passed through TokenSniffer.
A special member cached is set depending on whether the reply was cached or not.
- Parameters
chain_id (int) –
address (Union[str, eth_typing.evm.HexAddress]) –
- Return type
- get_diagnostics()[source]
Get a diagnostics message.
Use for logging what kind of data we have collected
Example output:
Token sniffer info is: TokenSniffer cache database /Users/moo/.cache/tradingstrategy/tokensniffer.sqlite summary: Entries: 195 Max score: 100 Min score: 0 Avg score: 56.6
- Returns
Multi-line human readable string
- Return type