import logging
import sys
from concurrent.futures import wait
from datetime import datetime as dt
from typing import List

import pandas as pd
from py_jftech import (
    component, autowired, get_config, filter_weekend, asynchronized,
    parse_date, workday_range, is_workday, prev_workday, format_date
)

from api import (
    RoboExecutor, Datum, AssetPool, PortfoliosBuilder,
    PortfoliosRisk, PortfoliosHolder, DataSync, RoboExportor, BacktestStep, RebalanceSignal
)

logger = logging.getLogger(__name__)


@component(bean_name='backtest')
class BacktestExecutor(RoboExecutor):

    @autowired
    def __init__(self, datum: Datum = None, pool: AssetPool = None,
                 syncs: List[DataSync] = None, export: RoboExportor = None,
                 builder: PortfoliosBuilder = None, hold: PortfoliosHolder = None,
                 signal: RebalanceSignal = None):
        self._datum = datum
        self._pool = pool
        self._builder = builder
        self._hold = hold
        self._syncs = syncs
        self._export = export
        self._signal = signal
        self._config = get_config(__name__)['backtest']

    @staticmethod
    def get_last_business_day(start_date, end_date):
        transfer_date = get_config('portfolios')['holder']['warehouse-transfer-date']
        # 生成日期范围并转换为DataFrame
        dates = pd.date_range(start_date, end_date, freq='MS', closed='right')
        dates = [pd.to_datetime(f"{date.year}-{date.month}-{transfer_date}") for date in dates]
        dates.insert(0, start_date)
        df = pd.DataFrame({'dates': dates})
        df['dates'] = df['dates'].apply(lambda x: prev_workday(x))
        result = []
        for i in range(0, len(df), get_config('portfolios')['holder']['warehouse-frequency']):
            result.append(df.iloc[i]['dates'])
        delta = workday_range(result[0], result[1])
        period = get_config(__name__)['backtest']['sealing-period']
        if len(delta) <= period:
            result.pop(1)
        return result

    @property
    def start_date(self):
        return pd.to_datetime(filter_weekend(self._config['start-date']))

    @property
    def start_step(self) -> BacktestStep:
        return BacktestStep(self._config['start-step'])

    @property
    def end_step(self) -> BacktestStep:
        return BacktestStep(self._config['end-step']) if 'end-step' in self._config else BacktestStep.HOLD_PORTFOLIO

    @property
    def is_sync_data(self):
        return get_config(__name__)['sync-data']

    @property
    def end_date(self):
        return pd.to_datetime(self._config['end-date'])

    @property
    def is_clean_up(self):
        return self._config['clean-up'] if 'clean-up' in self._config else True

    def clear_datas(self):
        if self.start_step.within(BacktestStep.ASSET_POOL) and self.end_step.without(BacktestStep.ASSET_POOL):
            logger.info('start to clear asset pool'.center(50, '-'))
            self._pool.clear()
        if self.start_step.within(BacktestStep.NORMAL_PORTFOLIO) and self.end_step.without(
                BacktestStep.NORMAL_PORTFOLIO):
            logger.info('start to clear normal portfolios'.center(50, '-'))
            self._builder.clear()
            self._signal.clear()
        if self.start_step.within(BacktestStep.HOLD_PORTFOLIO) and self.end_step.without(BacktestStep.HOLD_PORTFOLIO):
            logger.info('start to clear hold portfolios'.center(50, '-'))
            self._hold.clear()

    def start_exec(self):
        if self.is_sync_data:
            for sync in self._syncs:
                sync.do_sync()
        if self.is_clean_up:
            self.clear_datas()
        if self.start_step.within(BacktestStep.ASSET_POOL) and self.end_step.without(BacktestStep.ASSET_POOL):
            logger.info("start to build asset pool".center(50, '-'))
            now = dt.now()
            workdays = self.get_last_business_day(self.start_date, self.end_date)
            for date in workdays:
                self._pool.get_pool(date)
                logger.info(f"build asset pool success, use[{(dt.now() - now).seconds}s]")
        if self.start_step.within(BacktestStep.NORMAL_PORTFOLIO) and self.end_step.without(
                BacktestStep.NORMAL_PORTFOLIO):
            logger.info("start to build normal portfolios".center(50, '-'))
            now = dt.now()
            wait([self.async_build_portfolios(day, risk) for risk in PortfoliosRisk for day in
                  self.get_last_business_day(self.start_date, self.end_date)])
            logger.info(f"build normal portfolios success, use[{(dt.now() - now).seconds}s]")
        if self.start_step.within(BacktestStep.HOLD_PORTFOLIO) and self.end_step.without(BacktestStep.HOLD_PORTFOLIO):
            logger.info("start to build hold portfolios".center(50, '-'))
            now = dt.now()
            wait([self.async_build_hold(x) for x in PortfoliosRisk])
            logger.info(f"build hold portfolios success, use[{(dt.now() - now).seconds}s]")
        logger.info("start to export report".center(50, '-'))
        now = dt.now()
        self._export.export(max_date=self.end_date, min_date=self.start_date)
        logger.info(f"report file exported successfully. use[{(dt.now() - now).seconds}s].")

    @asynchronized(isolate=True)
    def async_build_risk_date(self, asset_id):
        self._risk.build_risk_date(asset_id, self.end_date)

    @asynchronized(isolate=True)
    def async_build_portfolios(self, day, risk: PortfoliosRisk):
        self._builder.get_portfolios(day, risk)

    @asynchronized(isolate=True)
    def async_build_hold(self, risk: PortfoliosRisk):
        self._hold.build_hold_portfolio(day=self.end_date, risk=risk)


@component(bean_name='real')
class RealExecutor(RoboExecutor):

    @autowired(names={'daily_export': 'daily-real-export'})
    def __init__(self, builder: PortfoliosBuilder = None, hold: PortfoliosHolder = None, syncs: List[DataSync] = None,
                 daily_export: RoboExportor = None, monitor_export: RoboExportor = None, pool: AssetPool = None,
                 signal: RebalanceSignal = None):
        self._builder = builder
        self._pool = pool
        self._hold = hold
        self._syncs = syncs
        self._daily_export = daily_export
        self._monitor_export = monitor_export
        self._config = get_config(__name__)['real']
        self._signal = signal

    @property
    def start_date(self):
        return pd.to_datetime(filter_weekend(self._config['start-date']))

    @property
    def is_sync_data(self):
        return get_config(__name__)['sync-data']

    @property
    def curt_date(self):
        if len(sys.argv) > 1:
            try:
                return parse_date(sys.argv[1])
            except Exception as e:
                logger.warning(f'get curt date from argv failure.', e)
        return dt.combine(dt.today().date(), dt.min.time())

    @property
    def include_date(self):
        return [dt.combine(x, dt.min.time()) for x in self._config['include-date']]

    @property
    def export(self):
        return self._config['export'] if 'export' in self._config else False

    def start_exec(self):
        if self.is_sync_data:
            for sync in self._syncs:
                sync.do_sync()
        date = self.curt_date
        if is_workday(date) or date in self.include_date:
            date = prev_workday(filter_weekend(date))
            for risk in PortfoliosRisk:
                logger.info(f"start to build risk[{risk.name}] real for date[{format_date(date)}]".center(50, '-'))
                now = dt.now()
                # 因为每天都必须有NORMAL最优投组,不管用不用
                self._builder.get_portfolios(date, risk)
                self._signal.get_signal(date, risk)
                # 更新持仓
                self._hold.build_hold_portfolio(date, risk)
                logger.info(
                    f"build risk[{risk.name}] real for date[{format_date(date)}] success, use[{(dt.now() - now).seconds}s]")
            if self.export:
                now = dt.now()
                # 每日实盘报告
                self._daily_export.export(max_date=date)
                logger.info(
                    f'export email for date[{format_date(date)}] send success, use[{(dt.now() - now).seconds}s]')
        else:
            logger.info(f'today[{format_date(date)}] is a rest day, do not execute the daily real robo.')