update

.gitignore

+### Python template
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+.idea
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/latest/usage/project/#working-with-version-control
+.pdm.toml
+.pdm-python
+.pdm-build/
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+.idea/
+

lib/test_api_agent.py

-import urllib3
-import json
-from datetime import datetime
-
-try:
-    from __main__ import _NOW__
-except:
-    _NOW__ = ''
-
-
-class Api_agent():
-
-    def __init__(self, portfolioDataPath="", assetClassPath=''):
-        if portfolioDataPath == "":
-            portfolioDataPath = 'test_data/portfolio.json'
-        if assetClassPath == "":
-            assetClassPath = 'test_data/class.json'
-        self.__DEBUG_INFO = ""
-        self.__classdata = json.load(open(assetClassPath, encoding='utf8'))
-        if type(portfolioDataPath) == str:
-            self.__portfolioData  = json.load(open(portfolioDataPath, encoding='utf8'))
-        else: 
-            self.__portfolioData  = json.load(open(portfolioDataPath[0], encoding='utf8'))
-            if portfolioDataPath[1] != "":
-                self.__portfolioData_record  = json.load(open(portfolioDataPath[1], encoding='utf8'))
-            
-        self.__portfolios = {}
-        self.__rp_idPortfolios = {}
-        self.today = datetime.now().strftime("%Y%m%d")
-        if _NOW__ != '':
-            self.today = _NOW__
-            print(f"日期非採用系統時間 ，採用傳入時間{self.today}")
-    '''    
-    def __get_request(self, query_string):
-        try:
-            rel = self.__TV_API.request("GET", query_string, headers = self.__HEADER_VALUES)
-            if rel == None:
-                pe
-                return None
-            data = ""
-            data = json.loads(rel.data.decode('utf-8'))
-            if len(data['data']) == 0:
-                return None
-            return data
-        except Exception as e:
-            self.__DEBUG_INFO = str(e)
-            return None
-    '''
-    def get_debug_info(self):
-        return self.__DEBUG_INFO
-
-    def get_latest_portfolio(self, portfolio_id, query_date):
-        query_date = query_date.replace('-','')
-        if query_date != '' and query_date != self.today:
-            self.__DEBUG_INFO = ('目前不提供指定日期取得推薦功能，除非是本日(%s)，實際輸入日期為(%s)' %(self.today,query_date ))
-            #print("測試時忽略推薦日期檢查")
-            print("***  測試時間不同，可能是 {vender}_quant_api.py 未加入 replace datetime.now()的程式碼")
-            return None
-        if portfolio_id in self.__portfolios.keys():
-            return self.__portfolios[portfolio_id]
-
-        try:
-            data = self.__portfolioData
-            if len(data['rps']) == 0:
-                return None
-
-            for d in data['rps'] :
-                if d['recomm_guid'] == portfolio_id:
-                    if not 'data_Date' in d:
-                        print(" Warning: `data_Date` is None, 但是不影響測試，只影響生產環境")
-                    elif  d['data_Date'] == self.today :
-                        pass
-                    else: 
-                        print(" Warning: `data_Date` is invalidated, 但是不影響測試，只影響生產環境")
-                        
-                    self.__portfolios[portfolio_id] = d
-                    return self.__portfolios[portfolio_id]
-                    
-
-            return None
-        except Exception as e:
-            self.__DEBUG_INFO = str(e)
-            return None
-        pass
-
-        api_query = "%s/franklin/prediction_data/%s/%s" % (self.__TV_API_URL, portfolio_id, query_date)
-        return self.__get_request(api_query)
-
-    def get_portfolio_by_rpid(self, rp_id):
-        rp_id = str(rp_id)
-        if rp_id in self.__rp_idPortfolios.keys():
-            return self.__rp_idPortfolios[rp_id]
-
-        try:
-            data = self.__portfolioData_record
-            if len(data['rps']) == 0:
-                return None
-
-            for d in data['rps'] :
-                d['rp_id'] = str(d['rp_id'])
-                if d['rp_id'] == rp_id:
-                    self.__rp_idPortfolios[rp_id] = d
-                    return self.__rp_idPortfolios[rp_id]
-
-            return None
-        except Exception as e:
-            self.__DEBUG_INFO = str(e)
-            return None
-        pass
-
-        api_query = "%s/franklin/prediction_data/%s/%s" % (self.__TV_API_URL, portfolio_id, query_date)
-        return self.__get_request(api_query)
-
-    def get_fund_metadata(self, fund_id):
-        return self.__classdata[fund_id]
-        pass
-
-        api_query = "%s/franklin/fundpool/%s" % (self.__TV_API_URL, fund_code)
-        data = self.__get_request(api_query)
-        if not data == None:
-            return data['data'][0]
-        return None

requirements.txt

@@ -7,7 +7,7 @@ Jinja2==2.10
 joblib==0.14.0
 MarkupSafe
 numpy==1.17.2
-pandas==0.25.1
+pandas
 python-dateutil==2.8.0
 pytz==2018.7
 scikit-learn==0.21.3

test_case_exec.txt

+生成测试用例：
+python test_case_gen\Sample\gentestcase.py test_case_gen\Sample\testcase_example_1114.xlsx
+
+
+执行测试用例：
+python reb_test.py sv test test_case_gen/Sample/TEST_CASE_1_portfolio.json test_ca
+se_gen/Sample/TEST_CASE_1_recomm.json test_data\class.json test_data\record_portfolio.json 20230804
+
+
+python reb_test.py sv test test_case_gen/Sample/TEST_CASE_2_portfolio.json test_ca
+se_gen/Sample/TEST_CASE_2_recomm.json test_data\class.json test_data\record_portfolio.json 20230810
+
+
+python reb_test.py sv test test_case_gen/Sample/TEST_CASE_3_portfolio.json test_ca
+se_gen/Sample/TEST_CASE_3_recomm.json test_data\class.json test_data\record_portfolio.json 20230817
+
+python reb_test.py sv test test_case_gen/Sample/TEST_CASE_4_portfolio.json test_case_gen/Sample/TEST_CASE_4_recomm.json test_data\class.json test_data\record_portfolio.json 20230812

test_case_gen/Sample/gentestcase.py

@@ -35,7 +35,7 @@ def main(xls_path):
 		obj['fas'] = fas
 		
 		#for i in range(10, len(df.loc[1:])):
-		for i in range(10, 20):
+		for i in range(11, 20):
 			try:
 				#print(str(df.loc[i][2]))
 				if str(df.loc[i][2]) == 'nan':
@@ -52,7 +52,7 @@ def main(xls_path):
 	
 		#print(json.dumps(portfolio))
 		outf = tc+'_portfolio.json'
-		print ( "create file: " + outf)
+		print("create file: " + outf)
 		with open(outf, 'w') as outfile:
 			json.dump(portfolio, outfile)
 
@@ -63,10 +63,11 @@ def main(xls_path):
 		rp['recomm_guid'] = str(df.loc[0][9])
 		rp['rp_id'] = df.loc[1][9]
 		rp['note'] = str(df.loc[8][2])
+		rp['data_Date'] = str(df.loc[9][2])
 		fws = []
 		rp['fws'] = fws
 		#for i in range(10, len(df.loc[1:])):
-		for i in range(10, 20):
+		for i in range(11, 21):
 			try:
 				#print(df.loc[i][8])
 				if str(df.loc[i][8]) == 'nan':
@@ -95,3 +96,4 @@ if __name__ == "__main__":
 	else:
 		usage()
 	
+

vender/sv_quant_api.py

+import calendar
+import json
+from datetime import datetime, timedelta
+
+from vender.quant_api_base import base_Quant_api
+
+
+class Quant_api(base_Quant_api):
+
+    def _rebalance(self, pt):
+
+        """
+        是否需要 rebalance
+        :param pt:
+        :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('last_check_date') or pt.get('last_check_date').replace('-', '') <= model_portfolio[
+                                                                                                  'data_Date']:
+            note = json.loads(pt.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')
+        return 0
+
+    def _dividend(self, pt):
+        # 投組不涉及配息
+        pass