跳转至

Config API 参考

qka.core.config

QKA 配置管理系统 提供统一的配置管理，支持环境变量、配置文件和代码配置

BacktestConfig dataclass

回测配置

Source code in qka/core/config.py 
13
14
15
16
17
18
19
20
21
@dataclass
class BacktestConfig:
    """回测配置"""
    initial_cash: float = 1_000_000  # 初始资金
    commission_rate: float = 0.0003  # 手续费率
    slippage: float = 0.001  # 滑点率
    min_trade_amount: int = 100  # 最小交易股数
    max_position_ratio: float = 0.3  # 单只股票最大仓位比例
    benchmark: str = '000300.SH'  # 基准指数

DataConfig dataclass

数据配置

Source code in qka/core/config.py 
24
25
26
27
28
29
30
31
32
@dataclass  
class DataConfig:
    """数据配置"""
    default_source: str = 'akshare'  # 默认数据源
    cache_enabled: bool = True  # 是否启用缓存
    cache_dir: str = './data_cache'  # 缓存目录
    cache_expire_days: int = 7  # 缓存过期天数
    quality_check: bool = True  # 是否进行数据质量检查
    auto_download: bool = True  # 是否自动下载缺失数据

TradingConfig dataclass

交易配置

Source code in qka/core/config.py 
35
36
37
38
39
40
41
42
43
@dataclass
class TradingConfig:
    """交易配置"""
    server_host: str = '0.0.0.0'
    server_port: int = 8000
    token_auto_generate: bool = True
    order_timeout: int = 30  # 订单超时时间(秒)
    max_retry_times: int = 3  # 最大重试次数
    heartbeat_interval: int = 30  # 心跳间隔(秒)

LogConfig dataclass

日志配置

Source code in qka/core/config.py 
46
47
48
49
50
51
52
53
54
55
@dataclass
class LogConfig:
    """日志配置"""
    level: str = 'INFO'
    console_output: bool = True
    file_output: bool = True
    log_dir: str = './logs'
    max_file_size: str = '10MB'
    backup_count: int = 10
    format: str = '[%(levelname)s][%(asctime)s][%(name)s] %(message)s'

PlotConfig dataclass

绘图配置

Source code in qka/core/config.py 
58
59
60
61
62
63
64
65
66
@dataclass
class PlotConfig:
    """绘图配置"""
    theme: str = 'plotly_white'
    figure_height: int = 600
    figure_width: int = 1000
    color_scheme: str = 'default'  # default, dark, colorful
    show_grid: bool = True
    auto_open: bool = True

Config

QKA 配置管理器

Source code in qka/core/config.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
class Config:
    """QKA 配置管理器"""

    def __init__(self, config_file: Optional[str] = None):
        """
        初始化配置管理器

        Args:
            config_file: 配置文件路径，为None时使用默认配置
        """
        # 默认配置
        self.backtest = BacktestConfig()
        self.data = DataConfig()
        self.trading = TradingConfig()
        self.log = LogConfig()
        self.plot = PlotConfig()

        # 加载配置文件
        if config_file and os.path.exists(config_file):
            self.load_from_file(config_file)

        # 加载环境变量配置
        self.load_from_env()

    def load_from_file(self, config_file: str):
        """从配置文件加载配置"""
        try:
            with open(config_file, 'r', encoding='utf-8') as f:
                config_data = json.load(f)

            # 更新各个配置段
            if 'backtest' in config_data:
                self._update_config(self.backtest, config_data['backtest'])
            if 'data' in config_data:
                self._update_config(self.data, config_data['data'])
            if 'trading' in config_data:
                self._update_config(self.trading, config_data['trading'])
            if 'log' in config_data:
                self._update_config(self.log, config_data['log'])
            if 'plot' in config_data:
                self._update_config(self.plot, config_data['plot'])

        except Exception as e:
            print(f"加载配置文件失败: {e}")

    def load_from_env(self):
        """从环境变量加载配置"""
        # 回测配置
        if os.getenv('QKA_INITIAL_CASH'):
            self.backtest.initial_cash = float(os.getenv('QKA_INITIAL_CASH'))
        if os.getenv('QKA_COMMISSION_RATE'):
            self.backtest.commission_rate = float(os.getenv('QKA_COMMISSION_RATE'))

        # 数据配置
        if os.getenv('QKA_DATA_SOURCE'):
            self.data.default_source = os.getenv('QKA_DATA_SOURCE')
        if os.getenv('QKA_CACHE_DIR'):
            self.data.cache_dir = os.getenv('QKA_CACHE_DIR')
          # 交易配置
        if os.getenv('QKA_SERVER_PORT'):
            self.trading.server_port = int(os.getenv('QKA_SERVER_PORT'))

    def _update_config(self, config_obj, config_dict: Dict[str, Any]):
        """更新配置对象"""
        for key, value in config_dict.items():
            if hasattr(config_obj, key):
                setattr(config_obj, key, value)

    def save_to_file(self, config_file: str):
        """保存配置到文件"""
        config_data = {
            'backtest': asdict(self.backtest),
            'data': asdict(self.data),
            'trading': asdict(self.trading),
            'log': asdict(self.log),
            'plot': asdict(self.plot)
        }

        # 确保目录存在（只有当目录路径不为空时）
        dir_path = os.path.dirname(config_file)
        if dir_path:
            os.makedirs(dir_path, exist_ok=True)

        with open(config_file, 'w', encoding='utf-8') as f:
            json.dump(config_data, f, indent=2, ensure_ascii=False)

    def to_dict(self) -> Dict[str, Any]:
        """转换为字典格式"""
        return {
            'backtest': asdict(self.backtest),
            'data': asdict(self.data),
            'trading': asdict(self.trading),
            'log': asdict(self.log),
            'plot': asdict(self.plot)
        }

    def get(self, section: str, key: str, default: Any = None) -> Any:
        """获取配置值"""
        section_obj = getattr(self, section, None)
        if section_obj:
            return getattr(section_obj, key, default)
        return default

    def set(self, section: str, key: str, value: Any):
        """设置配置值"""
        section_obj = getattr(self, section, None)
        if section_obj:
            setattr(section_obj, key, value)

__init__ 

__init__(config_file: Optional[str] = None)

初始化配置管理器

Parameters:

Name Type Description Default
config_file Optional[str]

配置文件路径，为None时使用默认配置

 None
Source code in qka/core/config.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def __init__(self, config_file: Optional[str] = None):
    """
    初始化配置管理器

    Args:
        config_file: 配置文件路径，为None时使用默认配置
    """
    # 默认配置
    self.backtest = BacktestConfig()
    self.data = DataConfig()
    self.trading = TradingConfig()
    self.log = LogConfig()
    self.plot = PlotConfig()

    # 加载配置文件
    if config_file and os.path.exists(config_file):
        self.load_from_file(config_file)

    # 加载环境变量配置
    self.load_from_env()

load_from_file 

load_from_file(config_file: str)

从配置文件加载配置

Source code in qka/core/config.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def load_from_file(self, config_file: str):
    """从配置文件加载配置"""
    try:
        with open(config_file, 'r', encoding='utf-8') as f:
            config_data = json.load(f)

        # 更新各个配置段
        if 'backtest' in config_data:
            self._update_config(self.backtest, config_data['backtest'])
        if 'data' in config_data:
            self._update_config(self.data, config_data['data'])
        if 'trading' in config_data:
            self._update_config(self.trading, config_data['trading'])
        if 'log' in config_data:
            self._update_config(self.log, config_data['log'])
        if 'plot' in config_data:
            self._update_config(self.plot, config_data['plot'])

    except Exception as e:
        print(f"加载配置文件失败: {e}")

load_from_env 

load_from_env()

从环境变量加载配置

Source code in qka/core/config.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def load_from_env(self):
    """从环境变量加载配置"""
    # 回测配置
    if os.getenv('QKA_INITIAL_CASH'):
        self.backtest.initial_cash = float(os.getenv('QKA_INITIAL_CASH'))
    if os.getenv('QKA_COMMISSION_RATE'):
        self.backtest.commission_rate = float(os.getenv('QKA_COMMISSION_RATE'))

    # 数据配置
    if os.getenv('QKA_DATA_SOURCE'):
        self.data.default_source = os.getenv('QKA_DATA_SOURCE')
    if os.getenv('QKA_CACHE_DIR'):
        self.data.cache_dir = os.getenv('QKA_CACHE_DIR')
      # 交易配置
    if os.getenv('QKA_SERVER_PORT'):
        self.trading.server_port = int(os.getenv('QKA_SERVER_PORT'))

save_to_file 

save_to_file(config_file: str)

保存配置到文件

Source code in qka/core/config.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def save_to_file(self, config_file: str):
    """保存配置到文件"""
    config_data = {
        'backtest': asdict(self.backtest),
        'data': asdict(self.data),
        'trading': asdict(self.trading),
        'log': asdict(self.log),
        'plot': asdict(self.plot)
    }

    # 确保目录存在（只有当目录路径不为空时）
    dir_path = os.path.dirname(config_file)
    if dir_path:
        os.makedirs(dir_path, exist_ok=True)

    with open(config_file, 'w', encoding='utf-8') as f:
        json.dump(config_data, f, indent=2, ensure_ascii=False)

to_dict 

to_dict() -> Dict[str, Any]

转换为字典格式

Source code in qka/core/config.py 
155
156
157
158
159
160
161
162
163
def to_dict(self) -> Dict[str, Any]:
    """转换为字典格式"""
    return {
        'backtest': asdict(self.backtest),
        'data': asdict(self.data),
        'trading': asdict(self.trading),
        'log': asdict(self.log),
        'plot': asdict(self.plot)
    }

get 

get(section: str, key: str, default: Any = None) -> Any

获取配置值

Source code in qka/core/config.py 
165
166
167
168
169
170
def get(self, section: str, key: str, default: Any = None) -> Any:
    """获取配置值"""
    section_obj = getattr(self, section, None)
    if section_obj:
        return getattr(section_obj, key, default)
    return default

set 

set(section: str, key: str, value: Any)

设置配置值

Source code in qka/core/config.py 
172
173
174
175
176
def set(self, section: str, key: str, value: Any):
    """设置配置值"""
    section_obj = getattr(self, section, None)
    if section_obj:
        setattr(section_obj, key, value)

load_config 

load_config(config_file: Optional[str] = None) -> Config

加载配置文件

Parameters:

Name Type Description Default
config_file Optional[str]

配置文件路径

 None

Returns:

Type Description
Config

Config对象

Examples:

使用默认配置

config = load_config()

从文件加载配置

config = load_config('config.json')

访问配置

print(config.backtest.initial_cash) print(config.data.default_source)

Source code in qka/core/config.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def load_config(config_file: Optional[str] = None) -> Config:
    """
    加载配置文件

    Args:
        config_file: 配置文件路径

    Returns:
        Config对象

    Examples:
        # 使用默认配置
        config = load_config()

        # 从文件加载配置
        config = load_config('config.json')

        # 访问配置
        print(config.backtest.initial_cash)
        print(config.data.default_source)
    """
    global config
    config = Config(config_file)
    return config

create_sample_config 

create_sample_config(file_path: str = 'qka_config.json')

创建示例配置文件

Parameters:

Name Type Description Default
file_path str

配置文件路径

 'qka_config.json'
Source code in qka/core/config.py 
209
210
211
212
213
214
215
216
217
218
def create_sample_config(file_path: str = 'qka_config.json'):
    """
    创建示例配置文件

    Args:
        file_path: 配置文件路径
    """
    sample_config = Config()
    sample_config.save_to_file(file_path)
    print(f"示例配置文件已创建: {file_path}")

配置示例

基本用法

from qka.core.config import Config

# 创建配置实例
config = Config()

# 从文件加载配置
config.load_from_file('config.yaml')

# 获取配置值
database_url = config.get('database.url', 'sqlite:///default.db')
debug_mode = config.get('debug', False)

# 设置配置值
config.set('api.timeout', 30)
config.set('logging.level', 'INFO')

环境变量配置

import os

# 设置环境变量
os.environ['QKA_DEBUG'] = 'true'
os.environ['QKA_DATABASE_URL'] = 'postgresql://localhost/qka'

# 加载环境变量配置
config = Config()
config.load_from_env()

# 访问配置
debug = config.get('debug')  # True
db_url = config.get('database.url')  # postgresql://localhost/qka

配置文件示例

YAML格式 (config.yaml)

# QKA量化系统配置文件

# 数据库配置
database:
  url: "sqlite:///qka.db"
  pool_size: 10
  echo: false

# API配置
api:
  host: "0.0.0.0"
  port: 8000
  timeout: 30
  rate_limit: 100

# 日志配置
logging:
  level: "INFO"
  format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
  file: "logs/qka.log"
  max_size: "10MB"
  backup_count: 5

# 交易配置
trading:
  commission: 0.001
  slippage: 0.0005
  initial_capital: 100000

# 风险管理
risk:
  max_position_size: 0.1
  max_drawdown: 0.2
  stop_loss: 0.05

# 策略配置
strategy:
  default_lookback: 252
  rebalance_frequency: "monthly"

# 通知配置
notifications:
  email:
    enabled: false
    smtp_server: "smtp.gmail.com"
    port: 587
  wechat:
    enabled: false
    webhook_url: ""

JSON格式 (config.json)

{
  "database": {
    "url": "sqlite:///qka.db",
    "pool_size": 10,
    "echo": false
  },
  "api": {
    "host": "0.0.0.0",
    "port": 8000,
    "timeout": 30
  },
  "logging": {
    "level": "INFO",
    "file": "logs/qka.log"
  }
}

配置验证

from qka.core.config import Config

config = Config()

# 定义配置验证规则
validation_rules = {
    'database.url': {'required': True, 'type': str},
    'api.port': {'required': True, 'type': int, 'min': 1, 'max': 65535},
    'trading.commission': {'type': float, 'min': 0, 'max': 1},
    'logging.level': {'type': str, 'choices': ['DEBUG', 'INFO', 'WARNING', 'ERROR']}
}

# 验证配置
try:
    config.validate(validation_rules)
    print("配置验证通过")
except ConfigError as e:
    print(f"配置验证失败: {e}")

动态配置更新

# 监听配置文件变化
config.watch_file('config.yaml', auto_reload=True)

# 注册配置变更回调
@config.on_change('database.url')
def on_database_change(old_value, new_value):
    print(f"数据库配置从 {old_value} 更改为 {new_value}")
    # 重新初始化数据库连接
    reconnect_database(new_value)

# 手动重新加载配置
config.reload()

配置模板生成

# 生成示例配置文件
config.create_sample_config('sample_config.yaml')

# 生成配置模板
template = config.get_config_template()
print(template)

最佳实践

  1. 配置文件管理
  2. 使用版本控制管理配置模板
  3. 敏感信息使用环境变量

  4. 不同环境使用不同配置文件

  5. 配置验证

  6. 应用启动时验证必需配置
  7. 定义清晰的验证规则

  8. 提供有意义的错误信息

  9. 配置更新

  10. 谨慎使用动态配置更新
  11. 关键配置变更需要重启服务

  12. 记录配置变更日志

  13. 安全考虑

  14. 敏感配置信息加密存储
  15. 限制配置文件访问权限
  16. 审计配置变更操作