Commit d4bbf72d authored by 吕先亚's avatar 吕先亚

refactor(sv_quant_api): 优化 rebalance 判断逻辑

-重构了 rebalance 函数,提高了代码可读性和维护性
- 新增 check_nan 函数,用于检查数据是否为空
- 新增 do_re_balance 函数,用于执行再平衡操作
- 优化了测试用例生成和执行的流程
parent 1e6d6d90
生成测试用例:
python test_case_gen\Sample\gentestcase.py test_case_gen\Sample\testcase_example.xlsx
python test_case_gen\Sample\gentestcase.py test_case_gen\Sample\testcase_example_0526.xlsx
执行测试用例:
......@@ -11,3 +11,5 @@ python reb_test.py sv test TEST_CASE_4_portfolio.json TEST_CASE_4_recomm.json te
python reb_test.py sv test TEST_CASE_5_portfolio.json TEST_CASE_5_recomm.json test_data\class.json test_data\record_portfolio.json 20230812
python reb_test.py sv test TEST_CASE_6_portfolio.json TEST_CASE_6_recomm.json test_data\class.json test_data\record_portfolio.json 20230812
python reb_test.py sv test TEST_CASE_7_portfolio.json TEST_CASE_7_recomm.json test_data\class.json test_data\record_portfolio.json 20230812
python reb_test.py sv test TEST_CASE_8_portfolio.json TEST_CASE_8_recomm.json test_data\class.json test_data\record_portfolio.json 20230812
python reb_test.py sv test TEST_CASE_10_portfolio.json TEST_CASE_10_recomm.json test_data\class.json test_data\record_portfolio.json 20230812
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):
......@@ -15,27 +51,19 @@ class Quant_api(base_Quant_api):
:return: 返回 model_portfolio 的 rp_id 表示需要 rebalance ,若是不需要 rebalance 返回 0
"""
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]
model_portfolio = self.get_latest_portfolio(pt['recomm_id'])
if not pt.get('rp_id') or pt.get('rp_id') != model_portfolio['rp_id']:
# 如果推薦生成日data_Date在最後一次觸發再平衡时间之后,视为需要再平衡
if (not pt.get('last_recomm_date') or type(pt.get('last_recomm_date')) != str or
model_portfolio['data_Date'] > pt.get('last_recomm_date').replace('-', '')):
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')
# 用户没买过投组
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):
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment