fix: 修复所有单测

lixinger_openapi/__init__.py

@@ -1,12 +1,11 @@
 # -*- coding: utf-8 -*-
-'''
+"""
 理杏仁开放平台API，非官方
-'''
+"""
 from lixinger_openapi.query import query_json, query_dataframe
 from lixinger_openapi.token import set_token
+from ._version import __version__
 
-__version__ = '1.0.2'
-
-__author__ = 'sheki lyu'
+__author__ = "sheki lyu"
 
 name = "lixinger_openapi"

lixinger_openapi/_version.py

@@ -0,0 +1 @@
+__version__ = '1.0.3'

lixinger_openapi/query.py

@@ -4,7 +4,7 @@
 '''
 import json
 import requests
-from pandas.io.json import json_normalize
+from pandas import json_normalize
 from lixinger_openapi.token import get_token
 
 BASEURL = "https://open.lixinger.com/api/"

requirement.txt

@@ -0,0 +1,6 @@
+numpy
+pandas
+requests
+setuptools
+tox
+pytest

setup.py

@@ -1,36 +1,57 @@
-# -*- coding: utf-8 -*-
-'''
+"""
 理杏仁开放平台API包安装
-'''
-import lixinger_openapi as lo
-from setuptools import (
-    find_packages,
-    setup,
-)
+"""
+
+import importlib.util
+from pathlib import Path
+from types import ModuleType
+
+from setuptools import find_packages, setup
+
+
+def get_version() -> str:
+    """
+    获取 _version.py 文件里的版本号
+    """
+    version_file: Path = Path(__file__).parent / "lixinger_openapi" / "_version.py"
+    assert version_file is not None, "版本文件不存在"
+    spec = importlib.util.spec_from_file_location("_version", version_file)
+    if spec is None or spec.loader is None:
+        print(f"Warning: Could not load version from {version_file}")
+        return ""
+
+    # 使用 importlib.util.module_from_spec 创建一个新的模块对象，并执行
+    version_module: ModuleType = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(version_module)
+    return version_module.__version__
 
-version = lo.__version__
 
-with open("README.md", "r") as fh:
+# 读取README.md内容作为长描述
+with open("README.md", "r", encoding="utf-8") as fh:
     long_description = fh.read()
 
 setup(
-    name='lixinger_openapi',
-    version=version,
-    description='lixinger openapi',
+    name="lixinger_openapi",
+    version=get_version(),
+    description="lixinger openapi",
     long_description=long_description,
     long_description_content_type="text/markdown",
     packages=find_packages(),
-    author='sheki lyu',
-    author_email='lvxueji@gmail.com',
-    license='Apache License v2',
+    author="sheki lyu",
+    author_email="lvxueji@gmail.com",
+    license="Apache License v2",
     install_requires=[
         "requests",
         "pandas",
+        "numpy",
     ],
-    url='https://github.com/ShekiLyu/lixinger-openapi',
+    url="https://github.com/ShekiLyu/lixinger-openapi",
     classifiers=[
-        'Operating System :: OS Independent',
-        'Programming Language :: Python :: 2.7',
-        'Programming Language :: Python :: 3',
+        "Operating System :: OS Independent",
+        "Programming Language :: Python :: 3",  # 移除对Python 2.7的支持
+        "License :: OSI Approved :: Apache Software License",
+        "Intended Audience :: Developers",
+        "Topic :: Software Development :: Libraries :: Python Modules",
     ],
+    python_requires=">=3.6",  # 指定最低Python版本要求
 )

test/query_test.py

@@ -1,84 +1,129 @@
 # -*- coding: utf-8 -*-
-'''
+"""
 测试代码
 
 注意：因为token值涉及账户隐私，不能在代码里写，所以请在代码所在目录下创建一个token.cfg文件，并将token值写入。token.cfg文件我已经在git里忽略，不会被上库。
-'''
-import sys, os
-if os.path.abspath('..') not in sys.path:
-    sys.path.append(os.path.abspath('..'))
+"""
+
+from datetime import datetime
+from pprint import pp
+import sys
+import os
 import unittest
 
-from lixinger_openapi.token import set_token
+import pandas as pd
+
+# 将上级目录添加到系统路径中，以便导入自定义模块
+if os.path.abspath("..") not in sys.path:
+    sys.path.append(os.path.abspath(".."))
 from lixinger_openapi.query import (
     query_json,
     query_dataframe,
 )
 
+
 class DataTest(unittest.TestCase):
+    """测试理杏仁开放平台API的数据查询功能"""
+
     def setUp(self):
+        """你可以在这里运行一次set_token，写入token.cfg文件，然后再删除token值，这样有了cfg文件以后都不用set_token了"""
         pass
-        #你可以在这里运行一次set_token，写入token.cfg文件，然后再删除token值，这样有了cfg文件以后都不用set_token了。
-        #set_token("")
-
-    def test_query_json_001(self):
-        rlt = query_json('a.stock', {
-            "industryType": "bank"
-        })
-        self.assertEqual(0, rlt['code'])
-        self.assertEqual("success", rlt['msg'])
-        self.assertEqual("000001", rlt['data'][0]['stockCode'])
-
-    def test_query_dataframe_001(self):
-        rlt = query_dataframe('a.stock', {
-            "industryType": "bank"
-        })
-        self.assertEqual(0, rlt['code'])
-        self.assertEqual("success", rlt['msg'])
-        self.assertEqual("000001", rlt['data'].loc[0, 'stockCode'])
-
-    def test_query_json_002(self):
-        rlt = query_json('a.stock.indice', {
-            "stockCode": "000028"
-        })
-        self.assertEqual(0, rlt['code'])
-        self.assertEqual("success", rlt['msg'])
-        self.assertEqual("cn", rlt['data'][0]['areaCode'])
-        self.assertEqual("399348", rlt['data'][0]['stockCode'])
-
-    def test_query_json_003(self):
-        rlt = query_json('a.stock.industry', {
-            "stockCode": "000028"
-        })
-        self.assertEqual(0, rlt['code'])
-        self.assertEqual("success", rlt['msg'])
-        self.assertEqual("cn", rlt['data'][0]['areaCode'])
-        self.assertEqual("C05", rlt['data'][0]['stockCode'])
-
-    def test_query_json_004(self):
-        rlt = query_json('a.indice.fundamental', {
-            "date": "2018-01-19",
-            "stockCodes": ["000016"],
-            "metrics": [
-                "pe_ttm.y_10.weightedAvg",
-                "pe_ttm.weightedAvg",
-                "mc"
-            ]
-        })
-        self.assertEqual(0, rlt['code'])
-        self.assertEqual("success", rlt['msg'])
-        self.assertEqual("2018-01-19", rlt['data'][0]['date'][:10])
-        self.assertLess(12, rlt['data'][0]['pe_ttm']['weightedAvg'])
-
-    def test_query_json_005(self):
-        rlt = query_json('a.indice.samples', {
-            "date": "2017-09-30",
-            "stockCodes": ["000016"]
-        })
-        self.assertEqual(0, rlt['code'])
-        self.assertEqual("success", rlt['msg'])
-        self.assertEqual("000016", rlt['data'][0]['stockCode'])
-        self.assertEqual("600000", rlt['data'][0]['samples'][0])
-
-if __name__ == '__main__':
+        # set_token("")
+
+    def test_query_json_company(self):
+        """测试查询大陆公司基础信息（JSON格式）"""
+        rlt = query_json("cn/company", {"fsTableType": "bank"})
+        pp(rlt)
+        self.assertIn("code", rlt)
+        self.assertEqual("success", rlt["message"])
+        self.assertGreater(len(rlt["data"]), 0)
+        self.assertIn("000001", [x["stockCode"] for x in rlt["data"]])
+
+    def test_query_df_company(self):
+        """测试查询大陆公司基础信息（DataFrame格式）"""
+        rlt: dict = query_dataframe("cn/company", {"fsTableType": "bank"})
+        pp(rlt)
+        self.assertIn("code", rlt)
+        self.assertGreater(len(rlt), 0)
+        df: pd.DataFrame = rlt["data"]
+        self.assertTrue(isinstance(df, pd.DataFrame))
+        self.assertTrue((df["stockCode"] == "000001").any())
+
+    def test_query_json_index(self):
+        """测试查询公司所属指数（以宁德时代为例，必在深圳成指）"""
+        rlt = query_json("cn/company/indices", {"stockCode": "300750"})
+        pp(rlt)
+        self.assertIn("code", rlt)
+        self.assertEqual("success", rlt["message"])
+        self.assertGreater(len(rlt), 0)
+        self.assertEqual("cn", rlt["data"][0]["areaCode"])
+        深圳成指 = "399001"
+        self.assertTrue(
+            any(
+                r
+                for r in rlt["data"]
+                if "stockCode" in r and r["stockCode"] == 深圳成指
+            )
+        )
+
+    def test_query_json_company_industries(self):
+        """测试查询公司所属行业（以宁德时代为例）"""
+        rlt = query_json("cn/company/industries", {"stockCode": "300750"})
+        pp(rlt)
+        self.assertIn("code", rlt)
+        self.assertEqual("success", rlt["message"])
+        self.assertGreater(len(rlt), 0)
+        self.assertEqual("cn", rlt["data"][0]["areaCode"])
+        self.assertTrue(
+            any(r for r in rlt["data"] if "stockCode" in r and r["stockCode"] == "C03")
+        )
+
+    def test_query_json_fundamental(self):
+        """测试查询指数基本面数据（以上证50指数为例）"""
+        date_to_test = "2024-12-10"
+        rlt = query_json(
+            "cn/index/fundamental",
+            {
+                "date": date_to_test,
+                "stockCodes": ["000016"],
+                "metricsList": ["pe_ttm.y10.mcw.cvpos", "pe_ttm.mcw", "mc"],
+            },
+        )
+        pp(rlt)
+        self.assertIn("code", rlt)
+        self.assertEqual("success", rlt["message"])
+        self.assertGreater(len(rlt["data"]), 0)
+        data = rlt["data"][0]
+        self.assertEqual(date_to_test, _extract_date(data["date"]))
+        self.assertAlmostEqual(10.75, rlt["data"][0]["pe_ttm.mcw"], delta=0.01)
+
+    def test_query_json_index_samples(self):
+        """测试查询指数样本（以上证50指数为例）"""
+        rlt = query_json(
+            "cn/index/constituents", {"date": "2017-09-30", "stockCodes": ["000016"]}
+        )
+        pp(rlt)
+        self.assertIn("code", rlt)
+        self.assertEqual("success", rlt["message"])
+        self.assertIn("000016", [x["stockCode"] for x in rlt["data"]])
+
+    def test_query_df_us_index(self):
+        """测试查询美股指数基本信息"""
+        rlt = query_dataframe("us/index", {})
+        pp(rlt)
+        self.assertIn("code", rlt)
+        self.assertEqual("", rlt["msg"])
+        df: pd.DataFrame = rlt["data"]
+        assert isinstance(df, pd.DataFrame)
+        self.assertTrue(isinstance(df, pd.DataFrame))
+        self.assertGreater(len(df), 0)
+        self.assertTrue((df["name"] == "标普500").any())
+
+
+def _extract_date(datetime_str: str) -> str:
+    """从ISO格式的日期时间字符串中提取日期部分"""
+    return datetime.fromisoformat(datetime_str).strftime("%Y-%m-%d")
+
+
+if __name__ == "__main__":
     unittest.main()

tox.ini

@@ -0,0 +1,10 @@
+[tox]
+envlist = py36, py37, py38, py39, py310, py311, py312, py313
+
+[testenv]
+deps =
+    pandas
+    requests
+    setuptools
+commands =
+    pytest