import calendar import json import math from datetime import datetime, timedelta from vender.quant_api_base import base_Quant_api def last_busi_date(day_str: str): year = int(day_str[:4]) month = int(day_str[4:6]) _, last_day = calendar.monthrange(year, month) last_day_date = datetime(year, month, last_day) last_day_date = last_day_date.replace(hour=23, minute=59, second=59, microsecond=997000) while last_day_date.weekday() >= 5: # weekday() 返回0-6,5是周六,6是周日 last_day_date -= timedelta(days=1) return last_day_date.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3] def check_nan(data): """ 检查输入数据是否为空 :param data: :return: """ if data is None: return True elif type(data) == float and math.isnan(data): return True elif type(data) == str and (data == 'nan' or data.strip() == ''): return True return False def do_re_balance(model_portfolio): note = json.loads(model_portfolio.get('note')) if not note.get('recomm_reason'): raise RuntimeError('未輸入推薦理由') return model_portfolio.get('rp_id'), '月初調倉日', last_busi_date( model_portfolio['data_Date']), note.get( 'recomm_reason') class Quant_api(base_Quant_api): def _rebalance(self, pt): """ 是否需要 rebalance :param pt: :return: 返回 model_portfolio 的 rp_id 表示需要 rebalance ,若是不需要 rebalance 返回 0 """ model_portfolio = self.get_latest_portfolio(pt['recomm_id']) # 用户没买过投组 if check_nan(pt.get('rp_id')): return do_re_balance(model_portfolio) else: # 买过的投组不是最新投组 if str(pt.get('rp_id')) != str(model_portfolio.get('rp_id')): if check_nan(pt.get('last_recomm_date')): return do_re_balance(model_portfolio) else: # 最后检查日不为空且投组的日期大于最后检查日 if model_portfolio['data_Date'] > pt.get('last_recomm_date').replace('-', ''): return do_re_balance(model_portfolio) return 0 def _dividend(self, pt): # 投組不涉及配息 pass