python_market_data_service/INTERNAL_INTERFACES.md

内部接口层文档

架构设计

对外接口 (AmazingDataAdapter)
    │
    ├── 内部接口层 (InternalDataService)
    │   ├── _MarketDataInternal (市场数据)
    │   ├── _BaseDataInternal (基础数据)
    │   └── _InfoDataInternal (信息数据)
    │
    └── AmazingData SDK
        ├── _market_data
        ├── _base_data
        └── _info_data

设计原则

1. 对外接口不直接调用 SDK - 所有 SDK 调用都通过内部接口层
2. 统一错误处理 - 内部接口层统一处理 SDK 异常
3. 日志记录 - 内部接口层统一记录调用日志
4. 便于测试 - 可以 mock 内部接口进行单元测试

内部接口类

_MarketDataInternal

封装 _market_data SDK 对象的方法：

方法	SDK 方法	说明
login()	_market_data.login()	登录
is_login()	_market_data.is_login()	检查登录状态
query_kline()	_market_data.query_kline()	查询K线
query_snapshot()	_market_data.query_snapshot()	查询快照

_BaseDataInternal

封装 _base_data SDK 对象的方法：

方法	SDK 方法	说明
get_code_list()	_base_data.get_code_list()	获取股票代码列表
get_future_code_list()	_base_data.get_future_code_list()	获取期货代码列表
get_code_info()	_base_data.get_code_info()	获取代码信息
get_calendar()	_base_data.get_calendar()	获取交易日历
get_adj_factor()	_base_data.get_adj_factor()	获取复权因子
get_backward_factor()	_base_data.get_backward_factor()	获取后复权因子
get_etf_pcf()	_base_data.get_etf_pcf()	获取ETF申赎数据
get_hist_code_list()	_base_data.get_hist_code_list()	获取历史代码列表

_InfoDataInternal

封装 _info_data SDK 对象的方法：

方法	SDK 方法	说明
get_equity_structure()	_info_data.get_equity_structure()	股本结构
get_share_holder()	_info_data.get_share_holder()	股东数据
get_holder_num()	_info_data.get_holder_num()	股东户数
get_income()	_info_data.get_income()	利润表
get_balance_sheet()	_info_data.get_balance_sheet()	资产负债表
get_cash_flow()	_info_data.get_cash_flow()	现金流量表
get_profit_express()	_info_data.get_profit_express()	业绩预告
get_profit_notice()	_info_data.get_profit_notice()	业绩快报
get_margin_summary()	_info_data.get_margin_summary()	融资融券汇总
get_margin_detail()	_info_data.get_margin_detail()	融资融券明细
get_long_hu_bang()	_info_data.get_long_hu_bang()	龙虎榜
get_block_trading()	_info_data.get_block_trading()	大宗交易
get_index_constituent()	_info_data.get_index_constituent()	指数成分股
get_index_weight()	_info_data.get_index_weight()	指数权重
get_fund_share()	_info_data.get_fund_share()	基金份额
get_kzz_issuance()	_info_data.get_kzz_issuance()	可转债发行
get_history_stock_status()	_info_data.get_history_stock_status()	历史股票状态（涨停/跌停/ST/停牌）

使用方式

在 AmazingDataAdapter 中

class AmazingDataAdapter(DataSourceAdapter):
    def __init__(self):
        # ... 其他初始化 ...
        self._internal: Optional[InternalDataService] = None
    
    async def connect(self, config: dict) -> None:
        # ... 登录和初始化 SDK ...
        
        # 初始化内部数据服务层
        self._internal = InternalDataService(self)
    
    def _fetch_klines_sync(self, ...):
        # 使用内部接口，而不是直接调用 SDK
        kline_dict = self._internal.market.query_kline(...)
        code_info = self._internal.base.get_code_info(...)
        equity = self._internal.info.get_equity_structure(...)

错误处理

内部接口统一处理 SDK 异常，返回空数据或默认值：

class _BaseDataInternal:
    def get_code_info(self, security_type: str) -> pd.DataFrame:
        try:
            return self._base_data.get_code_info(security_type=security_type)
        except Exception as e:
            error(f"[_BaseDataInternal] Get code info failed: {e}")
            return pd.DataFrame()  # 返回空数据而不是抛出异常

文件结构

app/adapters/
├── __init__.py
├── base.py                    # 适配器基类
├── amazingdata_adapter.py     # 对外适配器（使用内部接口）
└── internal_data_service.py   # 内部接口层

迁移说明

之前的调用方式（已废弃）

# 直接调用 SDK
code_info = self._base_data.get_code_info(security_type=...)
kline = self._market_data.query_kline(...)
equity = self._info_data.get_equity_structure(...)

现在的调用方式

# 通过内部接口调用
code_info = self._internal.base.get_code_info(security_type=...)
kline = self._internal.market.query_kline(...)
equity = self._internal.info.get_equity_structure(...)

优势

1. 解耦 - 对外接口与 SDK 解耦，便于更换数据源
2. 可测试性 - 可以 mock 内部接口进行单元测试
3. 错误处理 - 统一的错误处理和日志记录
4. 扩展性 - 可以在内部接口层添加缓存、限流等功能