mirror of
https://github.com/FlipsideCrypto/nft-deal-score.git
synced 2026-02-06 10:56:58 +00:00
90 lines
2.5 KiB
Python
90 lines
2.5 KiB
Python
import os
|
|
import re
|
|
import pandas as pd
|
|
import snowflake.connector
|
|
|
|
|
|
clean_names = {
|
|
'aurory': 'Aurory'
|
|
,'thugbirdz': 'Thugbirdz'
|
|
,'smb': 'Solana Monkey Business'
|
|
,'degenapes': 'Degen Apes'
|
|
,'peskypenguinclub': 'Pesky Penguins'
|
|
,'meerkatmillionaires': 'Meerkat Millionaires'
|
|
,'boryokudragonz': 'Boryoku Dragonz'
|
|
,'degods': 'DeGods'
|
|
,'lunabulls': 'LunaBulls'
|
|
,'boredapekennelclub': 'BAKC'
|
|
,'boredapeyachtclub': 'BAYC'
|
|
,'mutantapeyachtclub': 'MAYC'
|
|
,'bayc': 'BAYC'
|
|
,'bakc': 'BAKC'
|
|
,'mayc': 'MAYC'
|
|
,'solgods': 'SOLGods'
|
|
,'meerkatmillionairescc': 'Meerkat Millionaires'
|
|
,'ggsg:galacticgeckos': 'Galactic Geckos'
|
|
,'solstein': 'SolStein'
|
|
# ,'stonedapecrew': 'Stoned Ape Crew'
|
|
}
|
|
|
|
def get_ctx():
|
|
usr = os.getenv('SNOWFLAKE_USR')
|
|
pwd = os.getenv('SNOWFLAKE_PWD')
|
|
# with open('snowflake.pwd', 'r') as f:
|
|
# pwd = f.readlines()[0].strip()
|
|
# with open('snowflake.usr', 'r') as f:
|
|
# usr = f.readlines()[0].strip()
|
|
|
|
ctx = snowflake.connector.connect(
|
|
user=usr,
|
|
password=pwd,
|
|
account='vna27887.us-east-1'
|
|
)
|
|
return(ctx)
|
|
|
|
def format_num(x):
|
|
return('{:,}'.format(round(x, 2)))
|
|
|
|
def clean_token_id(df):
|
|
tokens = pd.read_csv('./data/tokens.csv')
|
|
df['collection'] = df.collection.apply(lambda x: clean_name(x))
|
|
df['token_id'] = df.token_id.apply(lambda x: re.sub('"', '', x) if type(x)==str else x )
|
|
df['tmp'] = df.token_id.apply(lambda x: x[:10] )
|
|
tokens['tmp'] = tokens.token_id.apply(lambda x: str(x)[:10] )
|
|
df = df.merge(tokens[['collection','tmp','clean_token_id']], how='left', on=['collection','tmp'])
|
|
df['token_id'] = df.clean_token_id.fillna(df.token_id)
|
|
df['token_id'] = df.token_id.astype(int)
|
|
del df['tmp']
|
|
del df['clean_token_id']
|
|
return(df)
|
|
|
|
def clean_name(name):
|
|
x = re.sub('-', '', name).lower()
|
|
x = re.sub(' ', '', x).lower()
|
|
if x in clean_names.keys():
|
|
return(clean_names[x])
|
|
name = name.title()
|
|
name = re.sub('-', ' ', name)
|
|
name = re.sub(' On ', ' on ', name)
|
|
name = re.sub('Defi ', 'DeFi ', name)
|
|
# name = re.sub(r'[^a-zA-Z0-9\s]', '', name)
|
|
return(name)
|
|
|
|
|
|
def merge(left, right, on=None, how='inner', ensure=True, verbose=True, message = ''):
|
|
df = left.merge(right, on=on, how=how)
|
|
if len(df) != len(left) and (ensure or verbose):
|
|
if message:
|
|
print(message)
|
|
print('{} -> {}'.format(len(left), len(df)))
|
|
cur = left.merge(right, on=on, how='left')
|
|
cols = set(right.columns).difference(set(left.columns))
|
|
# print(cols)
|
|
if ensure:
|
|
col = list(cols)[0]
|
|
missing = cur[cur[col].isnull()]
|
|
print(missing.head())
|
|
assert(False)
|
|
return(df)
|
|
|