%load_ext autoreload
%autoreload

from datetime import datetime
from signals.comps import CompCandidatesGenerator
from signals.comps.candidate_generator.dask_candidates_generator import DaskCompCandidatesGenerator
from signals.comps.constants import VALUATION_DATE_COLUMN
from signals.entities import ADDRESS_TOKEN
from signals.factory import create_app
from signals.markets import Market
from signals.dask.dask_client import start_dask_client
import pandas as pd
from dask import dataframe as dd
create_app('bla').app_context().push()
c = start_dask_client(safe=False)


market = Market.VEGAS
glavar_ats = ['0190c492-e0fe-53c6-bada-8d0097406b1f']
tts = [datetime(2016,8,8)]
ccg = CompCandidatesGenerator(market, comp_days_lookback=market.config['comp_days_lookback'])
ccg_df = ccg.fetch(glavar_ats, tts)

dccg = DaskCompCandidatesGenerator(
    market,
    asof_date=datetime(2017, 8, 7),
    read_env='production',
    comp_days_lookback=market.config['comp_days_lookback']
)
glavar_ddf = dd.from_pandas(
    pd.DataFrame({ADDRESS_TOKEN: glavar_ats, VALUATION_DATE_COLUMN: tts}), npartitions=1).set_index(
    ADDRESS_TOKEN, drop=False)
dccg_df = dccg.fetch(glavar_ddf).compute()
len(dccg_df)  # 499

set(ccg_df.to_records(index=False)) - set(dccg_df.to_records(index=False))
len(set([tuple(v) for v in ccg_df.values]) - set([tuple(v) for v in dccg_df.values])) # 0
len(set([tuple(v) for v in dccg_df.values]) - set([tuple(v) for v in ccg_df.values])) # 11
len(set([tuple(v) for v in ccg_df.values]))  # 499
len(set([tuple(v) for v in dccg_df.values]))  # 488
########################################################################################################################