quant/easyquant
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

init

Browse Source
This commit is contained in:
summer 2025-07-03 23:39:31 +08:00
commit 51ca0877d9
26 changed files with 3108 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored Normal file
View File

@ -0,0 +1 @@
.idea

0
config/__init__.py Normal file
View File

92
config/logs/app.log Normal file
View File

@ -0,0 +1,92 @@
2025-06-26 01:28:57 [INFO] 初始化策略配置
2025-06-26 01:28:59 [INFO] 注册策略序列:10000112
2025-06-26 01:28:59 [INFO] 确认配置参数:
2025-06-26 01:28:59 [INFO] 确认配置参数完毕, 如有错误请立刻停止执行
2025-06-26 01:33:36 [INFO] 初始化策略配置
2025-06-26 01:33:38 [INFO] 注册策略序列:10000114
2025-06-26 01:33:38 [INFO] 确认配置参数:
2025-06-26 01:33:38 [INFO] 参数名: stock_num, 值: 9, 备注: 买入股票的数量
2025-06-26 01:33:38 [INFO] 参数名: pass_month, 值: 1,4, 备注: 特殊月过滤, 1,4等于1月和4月
2025-06-26 01:33:38 [INFO] 参数名: key3, 值: new_value, 备注: None
2025-06-26 01:33:38 [INFO] 参数名: key4, 值: d, 备注: None
2025-06-26 01:33:38 [INFO] 参数名: example_key, 值: new_value, 备注: None
2025-06-26 01:33:38 [INFO] 参数名: ke3, 值: new_value, 备注: None
2025-06-26 01:33:38 [INFO] 参数名: run_stoploss, 值: True, 备注: None
2025-06-26 01:33:39 [INFO] 参数名: stoploss_strategy, 值: 3, 备注: None
2025-06-26 01:33:39 [INFO] 参数名: stoploss_limit, 值: 0.88, 备注: None
2025-06-26 01:33:39 [INFO] 参数名: stoploss_market, 值: 0.94, 备注: None
2025-06-26 01:33:39 [INFO] 参数名: index_stocks, 值: 399101.XSHE, 备注: None
2025-06-26 01:33:39 [INFO] 参数名: no_trading_today_signal, 值: True, 备注: None
2025-06-26 01:33:39 [INFO] 参数名: HV_control, 值: False, 备注: None
2025-06-26 01:33:39 [INFO] 参数名: HV_duration, 值: 120, 备注: None
2025-06-26 01:33:39 [INFO] 参数名: HV_ratio, 值: 0.9, 备注: None
2025-06-26 01:33:39 [INFO] 参数名: benchmark_stock, 值: 000001.XSHG, 备注: None
2025-06-26 01:33:39 [INFO] 参数名: pass_april, 值: True, 备注: None
2025-06-26 01:33:39 [INFO] 参数名: in_no_trading_period, 值: False, 备注: None
2025-06-26 01:33:39 [INFO] 确认配置参数完毕, 如有错误请立刻停止执行
2025-06-26 01:37:44 [INFO] 初始化策略配置
2025-06-26 01:37:44 [INFO] 注册策略序列:10000115
2025-06-26 01:37:44 [INFO] 确认配置参数:
2025-06-26 01:37:44 [INFO] 参数名: stock_num, 值: 9, 备注: 买入股票的数量
2025-06-26 01:37:44 [INFO] 参数名: pass_month, 值: 1,4, 备注: 特殊月过滤, 1,4等于1月和4月
2025-06-26 01:37:44 [INFO] 参数名: key3, 值: new_value, 备注: None
2025-06-26 01:37:44 [INFO] 参数名: key4, 值: d, 备注: None
2025-06-26 01:37:44 [INFO] 参数名: example_key, 值: new_value, 备注: None
2025-06-26 01:37:45 [INFO] 参数名: ke3, 值: new_value, 备注: None
2025-06-26 01:37:45 [INFO] 参数名: run_stoploss, 值: True, 备注: None
2025-06-26 01:37:45 [INFO] 参数名: stoploss_strategy, 值: 3, 备注: None
2025-06-26 01:37:45 [INFO] 参数名: stoploss_limit, 值: 0.88, 备注: None
2025-06-26 01:37:45 [INFO] 参数名: stoploss_market, 值: 0.94, 备注: None
2025-06-26 01:37:45 [INFO] 参数名: index_stocks, 值: 399101.XSHE, 备注: None
2025-06-26 01:37:45 [INFO] 参数名: no_trading_today_signal, 值: True, 备注: None
2025-06-26 01:37:45 [INFO] 参数名: HV_control, 值: False, 备注: None
2025-06-26 01:37:45 [INFO] 参数名: HV_duration, 值: 120, 备注: None
2025-06-26 01:37:45 [INFO] 参数名: HV_ratio, 值: 0.9, 备注: None
2025-06-26 01:37:45 [INFO] 参数名: benchmark_stock, 值: 000001.XSHG, 备注: None
2025-06-26 01:37:46 [INFO] 参数名: pass_april, 值: True, 备注: None
2025-06-26 01:37:46 [INFO] 参数名: in_no_trading_period, 值: False, 备注: None
2025-06-26 01:37:46 [INFO] 确认配置参数完毕, 如有错误请立刻停止执行
2025-06-26 01:38:26 [INFO] 初始化策略配置
2025-06-26 01:38:26 [INFO] 注册策略序列:10000116
2025-06-26 01:38:27 [INFO] 确认配置参数:
2025-06-26 01:38:27 [INFO] 参数名: stock_num, 值: 9, 备注: 买入股票的数量
2025-06-26 01:38:27 [INFO] 参数名: pass_month, 值: 1,4, 备注: 特殊月过滤, 1,4等于1月和4月
2025-06-26 01:38:27 [INFO] 参数名: key3, 值: new_value, 备注: None
2025-06-26 01:38:27 [INFO] 参数名: key4, 值: d, 备注: None
2025-06-26 01:38:27 [INFO] 参数名: example_key, 值: new_value, 备注: None
2025-06-26 01:38:27 [INFO] 参数名: ke3, 值: new_value, 备注: None
2025-06-26 01:38:27 [INFO] 参数名: run_stoploss, 值: True, 备注: None
2025-06-26 01:38:27 [INFO] 参数名: stoploss_strategy, 值: 3, 备注: None
2025-06-26 01:38:27 [INFO] 参数名: stoploss_limit, 值: 0.88, 备注: None
2025-06-26 01:38:27 [INFO] 参数名: stoploss_market, 值: 0.94, 备注: None
2025-06-26 01:38:27 [INFO] 参数名: index_stocks, 值: 399101.XSHE, 备注: None
2025-06-26 01:38:27 [INFO] 参数名: no_trading_today_signal, 值: True, 备注: None
2025-06-26 01:38:27 [INFO] 参数名: HV_control, 值: False, 备注: None
2025-06-26 01:38:28 [INFO] 参数名: HV_duration, 值: 120, 备注: None
2025-06-26 01:38:28 [INFO] 参数名: HV_ratio, 值: 0.9, 备注: None
2025-06-26 01:38:28 [INFO] 参数名: benchmark_stock, 值: 000001.XSHG, 备注: None
2025-06-26 01:38:28 [INFO] 参数名: pass_april, 值: True, 备注: None
2025-06-26 01:38:28 [INFO] 参数名: in_no_trading_period, 值: False, 备注: None
2025-06-26 01:38:28 [INFO] 确认配置参数完毕, 如有错误请立刻停止执行
2025-06-26 21:39:32 [INFO] 初始化策略配置
2025-06-26 21:39:32 [INFO] 注册策略序列:10000117
2025-06-26 21:42:11 [INFO] 确认配置参数:
2025-06-26 21:42:52 [INFO] 参数名: stock_num, 值: 9, 备注: 买入股票的数量
2025-06-26 21:42:52 [INFO] 参数名: pass_month, 值: 1,4, 备注: 特殊月过滤, 1,4等于1月和4月
2025-06-26 21:42:52 [INFO] 参数名: key3, 值: new_value, 备注: None
2025-06-26 21:42:52 [INFO] 参数名: key4, 值: d, 备注: None
2025-06-26 21:42:52 [INFO] 参数名: example_key, 值: new_value, 备注: None
2025-06-26 21:42:52 [INFO] 参数名: ke3, 值: new_value, 备注: None
2025-06-26 21:42:53 [INFO] 参数名: run_stoploss, 值: True, 备注: None
2025-06-26 21:42:53 [INFO] 参数名: stoploss_strategy, 值: 3, 备注: None
2025-06-26 21:42:53 [INFO] 参数名: stoploss_limit, 值: 0.88, 备注: None
2025-06-26 21:42:53 [INFO] 参数名: stoploss_market, 值: 0.94, 备注: None
2025-06-26 21:42:53 [INFO] 参数名: index_stocks, 值: 399101.XSHE, 备注: None
2025-06-26 21:42:53 [INFO] 参数名: no_trading_today_signal, 值: True, 备注: None
2025-06-26 21:42:53 [INFO] 参数名: HV_control, 值: False, 备注: None
2025-06-26 21:42:53 [INFO] 参数名: HV_duration, 值: 120, 备注: None
2025-06-26 21:42:53 [INFO] 参数名: HV_ratio, 值: 0.9, 备注: None
2025-06-26 21:42:53 [INFO] 参数名: benchmark_stock, 值: 000001.XSHG, 备注: None
2025-06-26 21:42:53 [INFO] 参数名: pass_april, 值: True, 备注: None
2025-06-26 21:42:53 [INFO] 参数名: in_no_trading_period, 值: False, 备注: None
2025-06-26 21:42:53 [INFO] 确认配置参数完毕, 如有错误请立刻停止执行

226
config/network_config.py Normal file
View File

@ -0,0 +1,226 @@
import datetime
import requests
class network_config:
def __init__(self):
"""
初始化网络配置
"""
self.BASE_URL = "http://101.34.79.70:8081" # 替换为你的 NocoDB 地址
API_TOKEN = "mgIUTo_BATQnVPE3czsHLOLLgcQmnVv1qs7gJTIO" # 在用户设置中创建
self.headers = {
"xc-token": API_TOKEN
}
self.project_table = {
"strategy_config": "mw1zi3dr5sy03em", # 查看项目URL获取
"strategy_base": "mtvgo6pbrlmbgjc",
"strategy_config_template": "m4nty4o5t6sxzts",
"strategy_logs": "m8ebfypvbfpk38p",
"today_state": "mosii8hqry8xx9b"
}
pass
def register_strategy(self, strategy_name: str, template_id: str = ""):
"""
注册策略
:param strategy_name: 策略名称
:return: None
"""
PROJECT_ID = self.project_table['strategy_base']
url = f"{self.BASE_URL}/api/v2/tables/{PROJECT_ID}/records"
data = {
"text": strategy_name,
"memo": template_id,
"create_date": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), # 使用当前时间
"modify_date": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), # 使用当前时间
}
response = requests.post(url, headers=self.headers, data=data)
response.raise_for_status() # 检查错误
result = response.json()
template_list = self.get_template_list(template_id)
for template in template_list:
self.insert_config(result["id"], template)
return result["id"]
def get_strategy_id(self, strategy_name: str):
"""
获取策略ID
:param strategy_name: 策略名称
:return: 策略ID
"""
PROJECT_ID = self.project_table['strategy_base']
response = requests.get(f"{self.BASE_URL}/api/v2/tables/{PROJECT_ID}/records", headers=self.headers)
response.raise_for_status()
data = response.json()
records = data["list"]
return next((str(record["id"]) for record in records if record.get("text") == strategy_name), None)
def get_template_list(self, strategy_id: str) -> any:
"""
获取网络配置参数
:param type:
:param strategy_id:
:param key: 配置参数的键
:return: 配置参数的值
"""
PROJECT_ID = self.project_table['strategy_config_template']
url = f"{self.BASE_URL}/api/v2/tables/{PROJECT_ID}/records"
params = dict(limit=100, offset=0, where=f"(template_id,eq,{strategy_id})")
try:
return requests.get(url, headers=self.headers, params=params).json()['list']
except Exception as e:
return []
def get_data_list(self, strategy_id: str, type: str) -> any:
"""
获取网络配置参数
:param type:
:param strategy_id:
:param key: 配置参数的键
:return: 配置参数的值
"""
# 这里可以添加获取配置逻辑
PROJECT_ID = self.project_table['strategy_config']
url = f"{self.BASE_URL}/api/v2/tables/{PROJECT_ID}/records"
params = dict(limit=100, offset=0, where=f"(strategy_id,eq,{strategy_id})")
try:
result = []
records = requests.get(url, headers=self.headers, params=params).json()['list']
for record in records:
if record.get("strategy_id") == str(strategy_id) and record.get("type") == type:
result.append(record)
return result
except Exception as e:
return []
def update_config(self, id: str, new_value: str):
"""
更新网络配置参数
:return:
"""
PROJECT_ID = self.project_table['strategy_config']
url = f"{self.BASE_URL}/api/v2/tables/{PROJECT_ID}/records"
new_value = {
"id": id,
"new_value": new_value,
"create_date": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
}
update_response = requests.patch(url, headers=self.headers, json=new_value)
update_response.raise_for_status()
return update_response.json()
def insert_config(self, strategy_id: str, template: dict):
"""
插入网络配置参数
"""
PROJECT_ID = self.project_table['strategy_config']
url = f"{self.BASE_URL}/api/v2/tables/{PROJECT_ID}/records"
data = {
"strategy_id": strategy_id,
"type": "config",
"name": template['name'],
"data_type": template['data_type'],
"data_param": template['data_param'],
"memo": template['memo'],
"new_value": template['value'],
"create_date": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
}
response = requests.post(url, headers=self.headers, data=data)
response.raise_for_status() # 检查错误
# def update_check(self, id: str, old_value: str, new_value: str):
# """
# 更新网络配置参数
#
# :return:
# """
# PROJECT_ID = self.project_table['strategy_config']
# url = f"{self.BASE_URL}/api/v2/tables/{PROJECT_ID}/records/"
# new_value = {
# "id": id,
# "old_value": old_value,
# "new_value": new_value,
# "create_date": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
# }
# update_response = requests.patch(url, headers=self.headers, json=new_value)
# update_response.raise_for_status()
# return update_response.json()
#
# def insert_check(self, strategy_id: str, key: str, value: str):
# """
# 插入网络配置参数
# """
# PROJECT_ID = self.project_table['strategy_config']
# url = f"{self.BASE_URL}/api/v2/tables/{PROJECT_ID}/records"
# data = {
# "strategy_id": strategy_id,
# "type": "check",
# "name": key,
# "new_value": value,
# "create_date": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), # 使用当前时间
#
# }
# response = requests.post(url, headers=self.headers, data=data)
# response.raise_for_status() # 检查错误
@staticmethod
def insert_logs(strategy_id: str, log_type: str, log_content: str):
"""
插入日志记录
:param strategy_id: 策略ID
:param log_type: 日志类型
:param log_content: 日志内容
:return: None
"""
PROJECT_ID = "m8ebfypvbfpk38p"
url = f"http://101.34.79.70:8081/api/v2/tables/{PROJECT_ID}/records"
data = {
"strategy_id": strategy_id,
"msg_type": log_type,
"message": log_content,
"create_date": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), # 使用当前时间
}
API_TOKEN = "mgIUTo_BATQnVPE3czsHLOLLgcQmnVv1qs7gJTIO" # 在用户设置中创建
headers = {
"xc-token": API_TOKEN
}
response = requests.post(url, headers=headers, data=data)
response.raise_for_status()
@staticmethod
def insert_state(strategy_id: str,state_str:str,trade_date:str):
"""
:param strategy_id:
:param state_str:
:param trade_date:
:return:
"""
PROJECT_ID = "mosii8hqry8xx9b"
url = f"http://101.34.79.70:8081/api/v2/tables/{PROJECT_ID}/records"
data = {
"strategy_id": strategy_id,
"trade_date": trade_date, # 使用当前时间
"state": state_str
}
API_TOKEN = "mgIUTo_BATQnVPE3czsHLOLLgcQmnVv1qs7gJTIO" # 在用户设置中创建
headers = {
"xc-token": API_TOKEN
}
response = requests.post(url, headers=headers, data=data)
response.raise_for_status()
# aaa = network_config()
# ddd = aaa.get_data_list("10000113","config")
# print(ddd)

79
config/strategy_config.py Normal file
View File

@ -0,0 +1,79 @@
from config.strategy_state import state
from kit.conf import conf
from kit.logger import Logger
from config.network_config import network_config
class strategy_config:
def __init__(self,
strategy_name: str,
strategy_template_id: str,
config_save: network_config,
log):
"""
基础必要初始化策略配置
"""
# 策略名
self.strategy_name = strategy_name
# 配置模版id
self.template_id = strategy_template_id
# 配置变更保存方式
self.connection = config_save
# 日志记录器
self.logger = Logger(log)
"""
实例化strategy_config类时，初始化必要的属性
"""
# 策略ID
self.strategy_id = None
# 配置对应主键
self.configkey_id = {}
# 配置信息
self.config = {}
# 核心缓存
self.strategy_state = state()
# 初始化配置文件
self.init_from_db()
def init_from_db(self):
self.strategy_id = self.connection.register_strategy(self.strategy_name, self.template_id)
self.logger.info("初始化策略配置", self.strategy_id)
self.logger.info(' 注册策略序列:' + str(self.strategy_id), self.strategy_id)
config_list = self.connection.get_data_list(self.strategy_id, 'config')
self.config = {item['name']: item for item in config_list}
self.configkey_id = {item['name']: item['id'] for item in config_list}
self.logger.info('确认配置参数:', self.strategy_id)
for config in config_list:
self.logger.info(f" 参数名: {config['name']}, 值: {config['new_value']}, 备注: {config['memo']}", self.strategy_id)
self.logger.info('确认配置参数完毕, 如有错误请立刻停止执行', self.strategy_id)
def get_all_configs(self) -> dict:
"""
获取所有策略配置参数
:return: 包含所有配置参数的字典
"""
return self.config
def set_config(self, key: str, value: str) -> None:
"""
通过字典语法设置配置项，并自动触发数据库更新
Args:
key (str): 配置项名称
value (Any): 配置项的新值
"""
id = self.configkey_id.get(key)
self.connection.update_config(id, value)
def get_config(self, key: str):
func = conf.config_type(key)
return func(key)
# if __name__ == "__main__":
# config_instance = strategy_config('aaa', '10000000', network_config(), None)

138
config/strategy_state.py Normal file
View File

@ -0,0 +1,138 @@
from datetime import datetime
class hold_data:
def __init__(self, data: dict):
self.stock = None
self.price = None
self.amount = None
self.cost = None
self.profit = None
self.profit_rate = None
class state:
def __init__(self):
self.data = {}
self.init()
# 初始化
def init(self):
self.data = {
"create_date": "",
"position_list": {},
"sell_request_list": {},
"buy_request_list": {},
"high_stock_list": {},
"low_stock_list": {},
"reason_to_sell": "",
"history_buy_list": {},
"history_sell_list": {},
}
def get_buy_requests(self):
return self.data["buy_request_list"]
def clear_buy_requests(self):
"""
清空买入请求列表
"""
self.data["buy_request_list"] = {}
def reset(self):
self.init()
def set_state_datetime(self, key: str, value: datetime) -> None:
"""
设置状态值
:param key: 状态键
:param value: 状态值
"""
self.data[key] = value
def set_hold_stock(self, stock: str, data: dict) -> None:
"""
设置持仓股票数据
:param stock: 股票代码
:param data: hold_data 对象
"""
hd = hold_data(data)
if not isinstance(hd, hold_data):
raise ValueError("data must be an instance of hold_data")
self.data["position_list"][stock] = hd
def set_high_stock(self, stock: str, data: dict) -> None:
"""
设置高位股票数据
:param stock: 股票代码
:param data: hold_data 对象
"""
hd = hold_data(data)
if not isinstance(hd, hold_data):
raise ValueError("data must be an instance of hold_data")
self.data["high_stock_list"][stock] = hd
def set_low_stock(self, stock: str, data: dict) -> None:
"""
设置低位股票数据
:param stock: 股票代码
:param data: hold_data 对象
"""
hd = hold_data(data)
if not isinstance(hd, hold_data):
raise ValueError("data must be an instance of hold_data")
self.data["low_stock_list"][stock] = hd
def set_sell_request(self, stock: str, data: dict, reason: str) -> None:
"""
设置卖出请求数据
:param stock: 股票代码
:param data: hold_data 对象
"""
hd = hold_data(data)
if not isinstance(hd, hold_data):
raise ValueError("data must be an instance of hold_data")
self.data["sell_request_list"][stock + "_" + reason] = hd
def set_buy_request(self, stock: str, data: dict, reason: str) -> None:
"""
设置买入请求数据
:param stock: 股票代码
:param data: hold_data 对象
"""
hd = hold_data(data)
if not isinstance(hd, hold_data):
raise ValueError("data must be an instance of hold_data")
self.data["buy_request_list"][stock + "_" + reason] = hd
def clear_buys_requests(self):
"""
清空买入请求列表
"""
self.data["buy_request_list"] = {}
def get_sell_reason(self):
return self.data["reason_to_sell"]
def set_sell_reason(self, reason: str):
"""
设置卖出原因
:param reason: 卖出原因
"""
self.data["reason_to_sell"] = reason
def get_sell_requests(self):
"""
获取卖出请求列表
:return: 卖出请求列表
"""
return self.data["sell_request_list"]
def get_yesterday_high_list(self):
"""
获取昨日涨停股票列表
:return: 昨日涨停股票列表
"""
return self.data["high_stock_list"]

108
jq_channel.py Normal file
View File

@ -0,0 +1,108 @@
from typing import Any, List
class exchange:
def close_positions(self, context: Any, position: Any):
"""
关闭所有持仓
"""
raise NotImplementedError("This method should be implemented by subclasses")
def buy_security(self, context: Any, target_list: List[str]) -> None:
"""
买入操作：对目标股票执行买入，下单资金均摊分配
"""
raise NotImplementedError("This method should be implemented by subclasses")
def open_position(self, security: str, value: float) -> bool:
"""
开仓操作：尝试买入指定股票
参数:
security: 股票代码
value: 分配给该股票的资金
返回:
若下单成功（部分或全部成交）返回 True，否则返回 False
"""
raise NotImplementedError("This method should be implemented by subclasses")
class jq_channel(exchange):
def __init__(self):
pass
def order_target_value_(self, stock: str, value: float) -> Any:
"""
封装 order_target_value 函数进行下单，同时记录中文日志和异常信息
参数:
security: 股票代码
value: 下单目标金额
返回:
下单后生成的订单对象；若失败返回 None
"""
try:
order = order_target_value(stock, value)
return order
except Exception as e:
return None
def close_positions(self, context: Any, position: Any):
"""
平仓操作：尽可能将指定股票仓位全部卖出
参数:
position: 持仓对象
返回:
若下单后订单全部成交返回 True，否则返回 False
"""
security = position.security
order = self.order_target_value_(security, 0)
if order is None:
# log.error(f"股票 {security} 下单失败，可能是API调用错误")
return False
if order.status == OrderStatus.held and order.filled == order.amount:
# log.info(f"股票 {security} 卖出订单完全成交")
return True
else:
# log.warning(f"股票 {security} 卖出订单部分成交或未成交，状态: {order.status}, 已成交: {order.filled}/{order.amount}")
return False
def buy_security(self, context: Any, target_list: List[str]) -> None:
"""
买入操作：对目标股票执行买入，下单资金均摊分配
参数:
context: 聚宽平台传入的交易上下文对象
target_list: 目标股票代码列表
"""
position_count = len(context.portfolio.positions)
target_num = len(target_list)
if target_num > position_count:
try:
value = context.portfolio.cash / (target_num - position_count)
except ZeroDivisionError as e:
# log.error(f"资金分摊时除零错误: {e}")
return
for stock in target_list:
if context.portfolio.positions[stock].total_amount == 0:
if self.open_position(stock, value):
# log.info(f"已买入股票 {stock}，分配资金 {value:.2f}")
# self.not_buy_again.append(stock)
if len(context.portfolio.positions) == target_num:
break
def open_position(self, security: str, value: float) -> bool:
"""
开仓操作：尝试买入指定股票
参数:
security: 股票代码
value: 分配给该股票的资金
返回:
若下单成功（部分或全部成交）返回 True，否则返回 False
"""
order = self.order_target_value_(security, value)
if order is not None and order.filled > 0:
return True
return False

0
kit/__init__.py Normal file
View File

39
kit/conf.py Normal file
View File

@ -0,0 +1,39 @@
class conf:
@staticmethod
def to_int(value: str, params, separator: str = '_'):
return int(value)
@staticmethod
def to_float(value: str, params, separator: str = '_'):
return float(value)
@staticmethod
def to_str(value: str, params, separator: str = '_'):
return str(value)
@staticmethod
def to_bool(value: str, params, separator: str = '_'):
return bool(value)
@staticmethod
def to_list(value: str, params, separator: str = '_'):
params_list = params.split(separator, -1)
return value.split(params_list)
@staticmethod
def config_type(value: str):
data = {
"int": conf.to_int,
"float": conf.to_float,
"str": conf.to_str,
"bool": conf.to_bool,
"list": conf.to_list,
}
return data[value]
ddd = conf.config_type()
func = ddd["int"]
print(func("123")) # 输出: 123

292
kit/kit.py Normal file
View File

@ -0,0 +1,292 @@
from datetime import timedelta
from typing import Any, Optional, List, Dict
import pandas as pd
class DataHelper:
"""
数据操作辅助类
封装了数据接口的调用，包括 get_price 与 history 函数，
并在内部捕获异常、输出中文错误日志，避免重复编写 try/except 代码。
"""
@staticmethod
def get_price_safe(
security: Any,
end_date: Any,
frequency: str,
fields: List[str],
count: int,
panel: bool = False,
skip_paused: bool = True,
fq: Optional[str] = None,
fill_paused: bool = False
) -> Optional[pd.DataFrame]:
"""
安全调用 get_price 数据接口
参数:
security: 单只股票代码或股票代码列表
end_date: 数据截止日期
frequency: 数据频率，如 "daily" 或 "1m"
fields: 需要获取的数据字段列表（例如 ['open', 'close']）
count: 请求数据的记录数
panel: 是否返回面板数据（默认为 False）
skip_paused: 是否跳过停牌股票（默认为 True）
fq: 复权方式（例如"pre"或"post"，默认 None）
fill_paused: 是否填充停牌数据（默认为 False）
返回:
返回包含数据的 DataFrame，如果出错则返回 None
"""
try:
# 调用聚宽提供的 get_price 获取数据
df = get_price(
security,
end_date=end_date,
frequency=frequency,
fields=fields,
count=count,
panel=panel,
skip_paused=skip_paused,
fq=fq,
fill_paused=fill_paused
)
return df
except Exception as e:
# 输出中文错误日志，并返回 None
# logger.error(f"获取 {security} 的价格数据时出错: {e}")
return None
@staticmethod
def get_current_data():
"""
获取当前市场数据
返回:
当前市场数据的字典，包含股票代码到数据对象的映射
"""
try:
# 调用聚宽提供的 get_current_data 获取当前数据
return get_current_data()
except Exception as e:
# Logger.error(f"获取当前市场数据时出错: {e}")
return {}
@staticmethod
def get_history_safe(
security: Any,
unit: str,
field: str,
count: int,
) -> Optional[Dict[str, List[float]]]:
"""
安全调用 history 数据接口，批量获取历史数据
参数:
security: 单只或多只股票代码
unit: 数据单位，例如 "1m" 表示1分钟数据
field: 请求数据字段名称，如 "close"（收盘价）
count: 请求历史数据记录数
返回:
返回一个字典，映射股票代码到对应的数据列表；出错则返回 None
"""
try:
# 调用聚宽的 history 函数获取数据
data = history(count, unit=unit, field=field, security_list=security)
return data
except Exception as e:
return None
@staticmethod
def get_index_stocks(index_code: str) -> List[str]:
"""
获取指定指数的成分股列表
参数:
index_code: 指数代码，例如 "000300.XSHG" 表示沪深300指数
date: 可选参数，指定查询日期，默认为 None，表示查询最新成分股
返回:
成分股代码列表
"""
try:
# 调用聚宽的 get_index_stocks 函数获取成分股
return get_index_stocks(index_code)
except Exception as e:
return []
@staticmethod
def get_security_info(security: str) -> Optional[Any]:
"""
获取指定股票的安全信息
参数:
security: 股票代码，例如 "000001.XSHE"
返回:
股票的安全信息对象，如果出错则返回 None
"""
try:
# 调用聚宽的 get_security_info 函数获取股票信息
return get_security_info(security)
except Exception as e:
return None
@staticmethod
def get_stock_list(context: Any, index_stocks: str, stock_num: int) -> List[str]:
"""
选股模块：
1. 从指定股票池（如 399101.XSHE 指数成分股）中获取初步股票列表；
2. 应用多个过滤器筛选股票（次新股、科创股、ST、停牌、涨跌停等）；
3. 基于基本面数据（EPS、市值）排序后返回候选股票列表。
参数:
context: 聚宽平台传入的交易上下文对象
返回:
筛选后的候选股票代码列表
"""
# 从指定指数中获取初步股票列表
initial_list: List[str] = DataHelper.get_index_stocks(index_stocks)
# 依次应用过滤器，筛去不符合条件的股票
initial_list = DataHelper.filter_new_stock(context, initial_list) # 过滤次新股
initial_list = DataHelper.filter_kcbj_stock(initial_list) # 过滤科创/北交股票
initial_list = DataHelper.filter_st_stock(initial_list) # 过滤ST或风险股票
initial_list = DataHelper.filter_paused_stock(initial_list) # 过滤停牌股票
initial_list = DataHelper.filter_limitup_stock(context, initial_list) # 过滤当日涨停（未持仓时）的股票
initial_list = DataHelper.filter_limitdown_stock(context, initial_list) # 过滤当日跌停（未持仓时）的股票
# 利用基本面查询获取股票代码和EPS数据，并按照市值升序排序
q = query(valuation.code, indicator.eps) \
.filter(valuation.code.in_(initial_list)) \
.order_by(valuation.market_cap.asc())
df = get_fundamentals(q)
stock_list: List[str] = list(df.code)
stock_list = stock_list[:50] # 限制数据规模，防止一次处理数据过大
# 取前2倍目标持仓股票数作为候选池
final_list: List[str] = stock_list[:2 * int(stock_num)]
# 查询并输出候选股票的财务信息（如财报日期、营业收入、EPS）
if final_list:
info_query = query(
valuation.code,
income.pubDate,
income.statDate,
income.operating_revenue,
indicator.eps
).filter(valuation.code.in_(final_list))
df_info = get_fundamentals(info_query)
# for _, row in df_info.iterrows():
# log.info(
# f"股票 {row['code']}：报告日期 {row.get('pubDate', 'N/A')}，统计日期 {row.get('statDate', 'N/A')}，营业收入 {row.get('operating_revenue', 'N/A')}，EPS {row.get('eps', 'N/A')}")
return final_list
@staticmethod
def filter_paused_stock(stock_list: List[str]) -> List[str]:
"""
过滤停牌的股票
参数:
stock_list: 待过滤的股票代码列表
返回:
未停牌的股票代码列表
"""
current_data = DataHelper.get_current_data()
return [stock for stock in stock_list if not current_data[stock].paused]
@staticmethod
def filter_st_stock(stock_list: List[str]) -> List[str]:
"""
过滤带有 ST 或其他风险标识的股票
参数:
stock_list: 待过滤的股票代码列表
返回:
无 ST 或风险标识的股票代码列表
"""
current_data = DataHelper.get_current_data()
return [stock for stock in stock_list if (not current_data[stock].is_st) and
('ST' not in current_data[stock].name) and
('*' not in current_data[stock].name) and
('退' not in current_data[stock].name)]
@staticmethod
def filter_kcbj_stock(stock_list: List[str]) -> List[str]:
"""
过滤科创、北交股票
参数:
stock_list: 待过滤的股票代码列表
返回:
过滤后的股票代码列表（排除以 '4'、'8' 开头以及以 '68' 起始的股票）
"""
return [stock for stock in stock_list if stock[0] not in ('4', '8') and not stock.startswith('68')]
@staticmethod
def filter_limitup_stock(context: Any, stock_list: List[str]) -> List[str]:
"""
过滤当天已经涨停的股票（若未持仓则过滤）
参数:
context: 交易上下文对象
stock_list: 待过滤的股票代码列表
返回:
过滤后的股票代码列表
"""
history_data = DataHelper.get_history_safe(stock_list, unit='1m', field='close', count=1)
current_data = DataHelper.get_current_data()
if history_data is None:
return stock_list
return [stock for stock in stock_list if stock in context.portfolio.positions.keys() or
(history_data.get(stock, [0])[-1] < current_data[stock].high_limit)]
@staticmethod
def filter_limitdown_stock(context: Any, stock_list: List[str]) -> List[str]:
"""
过滤当天已经跌停的股票（若未持仓则过滤）
参数:
context: 交易上下文对象
stock_list: 待过滤的股票代码列表
返回:
过滤后的股票代码列表
"""
history_data = DataHelper.get_history_safe(stock_list, unit='1m', field='close', count=1)
current_data = DataHelper.get_current_data()
if history_data is None:
return stock_list
return [stock for stock in stock_list if stock in context.portfolio.positions.keys() or
(history_data.get(stock, [float('inf')])[-1] > current_data[stock].low_limit)]
@staticmethod
def filter_new_stock(context: Any, stock_list: List[str]) -> List[str]:
"""
过滤次新股：排除上市时间不足375天的股票
参数:
context: 交易上下文对象
stock_list: 待过滤的股票代码列表
返回:
过滤后的股票代码列表
"""
yesterday = context.previous_date
return [stock for stock in stock_list if not (yesterday - DataHelper.get_security_info(stock).start_date < timedelta(days=375))]
@staticmethod
def filter_highprice_stock(self, context: Any, stock_list: List[str]) -> List[str]:
"""
过滤股价高于设定上限（up_price）的股票（非持仓股票参与过滤）
参数:
context: 交易上下文对象
stock_list: 待过滤的股票代码列表
返回:
过滤后的股票代码列表
"""
history_data = DataHelper.get_history_safe(stock_list, unit='1m', field='close', count=1)
if history_data is None:
return stock_list
return [stock for stock in stock_list if stock in context.portfolio.positions.keys() or
history_data.get(stock, [self.up_price + 1])[-1] <= self.up_price]
@staticmethod
def filter_not_buy_again(stock_list: List[str], not_buy_again) -> List[str]:
"""
过滤掉当日已买入的股票，避免重复下单
参数:
stock_list: 待过滤的股票代码列表
返回:
未买入的股票代码列表
"""
return [stock for stock in stock_list if stock not in not_buy_again]

119
kit/logger.py Normal file
View File

@ -0,0 +1,119 @@
import logging
import os
from datetime import datetime
from config.network_config import network_config
class Logger:
"""
简单的日志操作类，支持控制台和文件输出
可以使用现有的logger对象或创建新的实例
"""
_instance = None
def __new__(cls, existing_logger=None, *args, **kwargs):
if existing_logger is not None:
# 如果提供了现有的logger对象，创建新实例并使用该对象
instance = super().__new__(cls)
instance.logger = existing_logger
return instance
# 否则使用单例模式
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, existing_logger=None, log_file="logs/app.log"):
# 如果使用现有logger，直接返回
if existing_logger is not None:
return
# 如果已经初始化过，直接返回
if hasattr(self, 'logger'):
return
# 创建logs目录（如果不存在）
os.makedirs(os.path.dirname(log_file), exist_ok=True)
# 初始化logger
self.logger = logging.getLogger('AppLogger')
self.logger.setLevel(logging.DEBUG)
# 清除可能存在的处理器
self.logger.handlers.clear()
# 设置日志格式
formatter = logging.Formatter(
'%(asctime)s [%(levelname)s] %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
# 添加控制台处理器
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
self.logger.addHandler(console_handler)
# 添加文件处理器
file_handler = logging.FileHandler(log_file, encoding='utf-8')
file_handler.setFormatter(formatter)
self.logger.addHandler(file_handler)
def debug(self, message, strategy_id=None):
"""记录调试级别的日志"""
self.logger.debug(message)
try:
network_config.insert_logs(strategy_id, "debug", message)
except:
pass
def info(self, message, strategy_id=None):
"""记录信息级别的日志"""
self.logger.info(message)
try:
network_config.insert_logs(strategy_id, "info ", message)
except:
pass
def warning(self, message, strategy_id=None):
"""记录警告级别的日志"""
self.logger.warning(message)
try:
network_config.insert_logs(strategy_id, "warn", message)
except:
pass
def error(self, message, strategy_id=None):
"""记录错误级别的日志"""
self.logger.error(message)
try:
network_config.insert_logs(strategy_id, "error", message)
except:
pass
def critical(self, message, strategy_id=None):
"""记录严重错误级别的日志"""
self.logger.critical(message)
try:
network_config.insert_logs(strategy_id, "critical", message)
except:
pass
# # 使用示例
# if __name__ == "__main__":
# # 示例1：创建新的logger实例
# logger = Logger()
# logger.info("这是一个新创建的logger实例")
#
# # 示例2：使用现有的logger对象
# existing_logger = logging.getLogger('ExistingLogger')
# existing_logger.setLevel(logging.DEBUG)
# handler = logging.StreamHandler()
# handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
# existing_logger.addHandler(handler)
#
# logger2 = Logger(existing_logger=existing_logger)
# logger2.info("这是使用现有logger对象创建的实例")
#
# # 示例3：测试单例模式（不传入existing_logger时）
# logger3 = Logger()
# logger3.info("这条日志来自第三个logger实例（与第一个是同一个实例）")

48
main.py Normal file
View File

@ -0,0 +1,48 @@
from logging import log
from typing import Any
from config.network_config import network_config
from jq_channel import jq_channel
from strategy import trade_strategy
from task.check_state_before_task import check_state_before_task
strategy = trade_strategy("测试策略", 10000000, network_config(), log, jq_channel())
def initialize(context: Any) -> None:
# # 初始化策略参数
strategy.init_strategy() # 策略初始化函数
# # 注册调度任务
# run_daily(check_positions_before_market_open_func, time='9:01') # 开盘前先检查持仓状态
# run_daily(check_holdings_yesterday_func, time='9:00')
# run_daily(prepare_stock_list_func, time='9:05')
#
# run_daily(sell_stocks_func, time='10:00') # 每日检查止损条件
#
# run_daily(process_pending_sells_func, time='10:15') # 处理待卖出股票
# run_weekly(buy_stocks_func, 2, time='10:30') # 周二进行调仓
# run_daily(process_pending_buys_func, time='10:20') # 处理待买入股票
# run_daily(process_pending_sells_func, time='13:05') # 午盘开盘后也处理一次卖出
# run_daily(process_pending_buys_func, time='13:15') # 午盘开盘后也处理一次买入
# run_daily(trade_afternoon_func, time='14:30')
# run_daily(process_pending_sells_func, time='14:40') # 收盘前再处理一次卖出
# run_daily(process_pending_buys_func, time='14:45') # 收盘前再处理一次买入
# run_daily(close_account_func, time='14:50')
# run_weekly(print_position_info_func, 5, time='15:10') # 周五收盘后打印持仓信息
task = check_state_before_task(strategy)
task.process(context)
pass
# initialize(None) # 调用初始化函数，开始策略执行

94
strategy.py Normal file
View File

@ -0,0 +1,94 @@
from config.strategy_state import state
from jq_channel import exchange
from kit.logger import Logger
from config.strategy_config import strategy_config
class trade_strategy:
def __init__(self, strategy_name, strategy_template_id, config_save, log, channel: exchange):
# -------必要参数-------
self.log = Logger(log) # 日志记录器
self.strategy_config = strategy_config(strategy_name,
strategy_template_id,
config_save,
self.log)
self.state = state() # 策略状态管理器，跟踪策略运行状态
self.channel = channel
# 添加持仓监控和管理器
def init_strategy(self) -> None:
"""
策略初始化函数
配置交易环境参数，包括防未来数据、基准、滑点、订单成本以及日志输出等级。
参数:
context: 聚宽平台传入的交易上下文对象
"""
# 启用防未来数据以确保历史回测的严谨性
self.log.info("初始化策略TradingStrategy配置...")
set_option('avoid_future_data', True)
# 设置策略基准为上证指数
#set_benchmark(self.strategy_config.config["benchmark_stock"])
set_benchmark("000001.XSHG")
# 使用真实市场价格，模拟实际交易
set_option('use_real_price', True)
# 设置固定滑点，确保价格偏差不大
set_slippage(FixedSlippage(3 / 10000))
# 设置订单成本，包括印花税和佣金
set_order_cost(OrderCost(
open_tax=0,
close_tax=0.001, # 卖出时0.1%的印花税
open_commission=2.5 / 10000,
close_commission=2.5 / 10000,
close_today_commission=0,
min_commission=5 # 最低佣金为5元
), type='stock')
# 设置日志输出等级（中文日志输出）
self.log.logger.set_level('order', 'error')
self.log.logger.set_level('system', 'error')
self.log.logger.set_level('strategy', 'debug')
def set_target_list(self, target_list):
self.target_list = target_list.copy()
def get_target_list(self):
"""获取本次调仓候选股票列表"""
return self.target_list
def get_configs(self) -> dict:
"""
获取策略配置
:return: strategy_config 对象
"""
return self.strategy_config.get_all_configs()
def get_strategy_config(self) -> strategy_config:
"""
获取策略配置字典
:return: 包含所有配置参数的字典
"""
return self.strategy_config
def close_position(self, context, stock: str):
"""
平仓操作：尽可能将指定股票仓位全部卖出
参数:
position: 持仓对象
返回:
若下单后订单全部成交返回 True，否则返回 False
"""
self.channel.close_positions(context, stock)
def open_position(self, security: str, value: float) -> bool:
"""
开仓操作：尝试买入指定股票
参数:
security: 股票代码
value: 分配给该股票的资金
返回:
若下单成功（部分或全部成交）返回 True，否则返回 False
"""
return self.channel.open_position(security, value)

0
task/__init__.py Normal file
View File

46
task/base_task.py Normal file
View File

@ -0,0 +1,46 @@
from typing import Any
from kit.conf import conf
from strategy import trade_strategy
class base_task:
def __init__(self, strategy: trade_strategy, sub_task=None):
self.strategy = strategy
self.name = "base_task"
self.remark = "base_task"
self.memo = "base_task"
self.sub_task = sub_task
self.strategy_config = self.strategy.get_strategy_config()
self.configs = self.strategy.get_configs()
self.log = self.strategy.get_logger()
def init(self, context: Any):
pass
def config(self, context: Any):
pass
def run(self, context: Any):
pass
def handle(self, context: Any):
pass
def end(self, context: Any):
pass
def process(self, context: Any):
self.init(context)
self.config(context)
self.run(context)
self.handle(context)
self.end(context)
def get_config(self, key: str):
configs = self.configs.get(key)
data_type = configs['data_type']
data_params = configs['data_params']
func = conf.config_type(data_type)
return func(self.configs[key]['new_value'], data_params)

73
task/buy_stocks_func_task.py Normal file
View File

@ -0,0 +1,73 @@
from typing import Any, List
from kit.kit import DataHelper
from strategy import trade_strategy
from task.base_task import base_task
class buy_stocks_func_task(base_task):
"""
每周调仓策略：
如果非空仓日，先选股得到目标股票列表，再卖出当前持仓中不在目标列表且昨日未涨停的股票，
最后买入目标股票，同时记录当天买入情况避免重复下单。
参数:
context: 聚宽平台传入的交易上下文对象
"""
def __init__(self, strategy: trade_strategy):
super().__init__(strategy)
def config(self, context: Any):
# 每次调仓的目标股票数量
self.stock_num = self.get_config('stock_num')
# 获取当前持仓列表
self.hold_list = [position.security for position in list(context.portfolio.positions.values())]
self.index_stocks = self.get_config('index_stocks')
self.target_list = DataHelper.get_stock_list(context, self.index_stocks, self.stock_num)
self.temp_target_list = self.target_list[:self.stock_num]
# 昨日涨停股票列表
self.yesterday_HL_list: List[str] = self.strategy.state.get_yesterday_high_list()
self.today_trade_switch = self.get_config('today_trade_switch')
def run(self, context: Any):
if not self.today_trade_switch:
self.log.info("今日非交易日，跳过调仓操作。")
return
# 重置当天已买入记录
# self.position_manager.reset_not_buy_again()
# 取目标持仓数以内的股票作为调仓目标
self.log.info(f"每周调仓目标股票: {self.temp_target_list}")
# 遍历当前持仓，若股票不在目标列表且非昨日涨停，则执行卖出操作
for stock in self.hold_list:
if stock not in self.temp_target_list and stock not in self.yesterday_HL_list:
pos = context.portfolio.positions.get(stock)
self.log.info(f"调仓决策：卖出股票 {stock}")
# 通过持仓监控器注册卖出请求，而不是直接卖出
self.strategy.state.set_sell_request(stock, pos, 'rebalance')
else:
self.log.info(f"调仓决策：继续持有股票 {stock}")
# 对目标股票执行买入操作前，先将其注册到待买入队列
buy_targets = [stock for stock in self.temp_target_list if stock not in self.hold_list]
if buy_targets:
for stock in buy_targets:
pos = context.portfolio.positions.get(stock)
self.strategy.state.set_buy_request(stock, pos, 'buy')
self.log.info(f"调仓决策：将 {len(buy_targets)} 只股票加入待买入队列: {buy_targets}")
# 更新当天已买入记录，防止重复买入
# for position in list(context.portfolio.positions.values()):
# if position.security not in self.position_manager.get_not_buy_again():
# self.position_manager.get_not_buy_again().append(position.security)
def handle(self, context: Any):
self.strategy.set_target_list(self.temp_target_list)
pass
def end(self, context: Any):
pass
# 这里可以添加任何必要的清理或总结操作

59
task/check_high_volume_func_task.py Normal file
View File

@ -0,0 +1,59 @@
from base.base_task import base_task
from kit.kit import DataHelper
from task.process_pending_sells_func_task import process_pending_sells_func_task
class check_high_volume_func_task(base_task):
"""
检查持仓股票当日成交量是否异常放量：
如果当日成交量大于过去 HV_duration 天内最大成交量的 HV_ratio 倍，则视为异常，执行卖出操作。
参数:
context: 聚宽平台传入的交易上下文对象
"""
def __init__(self, strategy, sub_task=None):
super().__init__(strategy, sub_task)
def init(self, context):
self.name = "check_high_volume_func_task"
self.remark = "检查高成交量任务"
self.memo = "检查是否有异常成交量的股票"
def begin(self, context):
self.HV_control = bool(self.config['HV_control'])
self.HV_duration = int(self.config['HV_duration'])
self.HV_ratio = float(self.config['HV_ratio'])
def run(self, context):
current_data = DataHelper.get_current_data()
for stock in list(context.portfolio.positions.keys()):
if current_data[stock].paused:
continue
if current_data[stock].last_price == current_data[stock].high_limit:
continue
if context.portfolio.positions[stock].closeable_amount == 0:
continue
df_volume = get_bars(
stock,
count=self.HV_duration,
unit='1d',
fields=['volume'],
include_now=True,
df=True
)
if df_volume is not None and not df_volume.empty:
if df_volume['volume'].iloc[-1] > self.HV_ratio * df_volume['volume'].max():
self.log.info(f"检测到股票 {stock} 出现异常放量，执行卖出操作。")
# 通过持仓监控器注册卖出请求，而不是直接卖出
self.position_monitor.register_sell_request(stock, 'high_volume')
# 处理所有待卖出请求
task = process_pending_sells_func_task(self.strategy, self.name)
task.process(context)
def handle(self, context):
pass
def end(self, context):
pass

58
task/check_limit_up_func_task.py Normal file
View File

@ -0,0 +1,58 @@
from typing import Any
from kit.kit import DataHelper
from task.base_task import base_task
class check_limit_up_func_task(base_task):
"""
检查昨日处于涨停状态的股票在当前是否破板。
如破板（当前价格低于涨停价），则立即卖出该股票，并记录卖出原因为 "limitup"。
参数:
context: 聚宽平台传入的交易上下文对象
"""
def __init__(self, strategy, sub_task=None):
super().__init__(strategy, sub_task)
def init(self, context: Any):
self.name = "check_limit_up_task"
self.remark = "检查涨停任务"
self.memo = "检查是否有涨停破板的股票"
def begin(self, context: Any):
self.yesterday_HL_list = self.strategy.state.get_yesterday_high_list()
def run(self, context: Any):
now_time = context.current_dt # 获取当前时间
if self.yesterday_HL_list:
for stock in self.yesterday_HL_list:
# 使用1分钟周期数据判断当前价格和涨停价是否符合条件
current_data = DataHelper.get_price_safe(
stock,
end_date=now_time,
frequency='1m',
fields=['close', 'high_limit'],
count=1,
panel=False,
fill_paused=True
)
if current_data is not None and not current_data.empty:
if current_data.iloc[0]['close'] < current_data.iloc[0]['high_limit']:
self.log.info(f"股票 {stock} 涨停破板，触发卖出操作。")
# 通过持仓监控器注册卖出请求，而不是直接卖出
self.strategy.state.set_sell_request(stock, current_data.iloc[0], 'limitup')
else:
self.log.info(f"股票 {stock} 仍维持涨停状态。")
# 处理待卖出股票
# if self.reason_to_sell[0] == 'limitup':
# sells_task = process_pending_sells_func_task(self.strategy, self.name)
# sells_task.process(context)
def handle(self, context: Any):
pass
def end(self, context: Any):
pass

72
task/check_market_env_task.py Normal file
View File

@ -0,0 +1,72 @@
from typing import Any
from strategy import trade_strategy
from task.base_task import base_task
class check_market_env_task(base_task):
"""
交易环境确认
1、判断是否为特殊月份（1月/4月）
2、如果是特殊月份且通过了清仓检查，则清仓所有持仓
3、如果不是特殊月份且没有交易信号，则重置交易信号
"""
def __init__(self, strategy: trade_strategy):
super().__init__(strategy)
def config(self, context: Any):
self.current_date = context.current_dt.date()
self.current_month = self.current_date.month
self.current_day = self.current_date.day
# 判断是否执行空仓策略
self.filter_month_clear_switch = self.get_config('filter_month_clear_switch')
self.filer_month_data = self.get_config('filter_month_data')
# 判断是否为特殊月 = True时 等于 执行1月/4月 全部清仓
self.in_special_month = str(self.current_month) in self.filer_month_data
self.today_trade_switch = self.get_config('today_trade_switch') # 是否为不交易信号
def run(self, context: Any):
#
if self.filter_month_clear_switch and self.in_special_month:
# 进入特殊月份，应清仓
for stock in self.strategy.state.data['position_list'].keys():
self.strategy.state.set_sell_request(stock, self.strategy.state.data['position_list'][stock], f"enter_{self.current_month}_month")
# 清仓后就不交易
self.today_trade_switch = False
elif not self.filter_month_clear_switch:
# 不在特殊月 或 不执行空仓策略 可以交易
self.today_trade_switch = True
else:
# 如果是特殊月份但没有清仓条件，则继续交易
self.today_trade_switch = True
def handle(self, context: Any):
# 保存状态
old_today_trade_switch = self.strategy_config.get_config("today_trade_switch")
if old_today_trade_switch != self.today_trade_switch:
self.strategy_config.set_config("today_trade_switch", self.today_trade_switch) # 临时变量
self.log.info(f"交易环境更新: today_trade_switch 状态更新为 {self.today_trade_switch}")
def end(self, context: Any):
self.log.info("-----------------")
self.log.info("一.今日环境确认:")
self.log.info(" 当前日期: {}".format(self.current_date))
self.log.info(" 空仓策略: {}".format("开启" if self.filter_month_clear_switch else "关闭"))
self.log.info(" 今日交易: {}".format("交易" if self.today_trade_switch else "清仓"))
self.log.info(" 特殊月份: {}".format(",".join(str(item) for item in self.filer_month_data)))
self.log.info(" 当前月份: {}".format(self.current_month))
# 今日清仓

67
task/check_positions_stop_loss_task.py Normal file
View File

@ -0,0 +1,67 @@
from typing import Any
from kit.kit import DataHelper
from strategy import trade_strategy
from task.base_task import base_task
class check_positions_stop_loss_task(base_task):
def __init__(self, strategy: trade_strategy):
super().__init__(strategy)
def init(self, context: Any):
pass
def config(self, context: Any):
self.stoploss_switch = self.get_config('stoploss_switch') # 是否启用止损策略
self.hold_list = self.strategy.state.data['position_list'] # 持仓列表
self.today_trade_switch = self.get_config('today_trade_switch') # 今日交易开关
self.stoploss_strategy = self.get_config('stoploss_strategy') # 止损策略
self.stoploss_stock_rate_data = self.get_config('stoploss_stock_rate_data') # 个股止损阀值
self.stoploss_market_rate_data = self.get_config('stoploss_market_rate_data') # 市场止损阀值
self.reason_to_sell = self.strategy.state.get_sell_reason() # 卖出原因
self.index_stocks = self.get_config('index_stocks') # 指数代码
def run(self, context: Any):
if len(self.hold_list) == 0:
self.log.debug("当前无持仓，跳过止损检查。")
return
if self.stoploss_switch and self.today_trade_switch:
if self.stoploss_strategy == 1 or self.stoploss_strategy == 3:
# 个股止盈或止损判断
for stock in list(context.portfolio.positions.keys()):
pos = context.portfolio.positions[stock]
if pos.price >= pos.avg_cost * 2:
# 通过持仓监控器注册卖出请求，而不是直接卖出
self.strategy.state.set_sell_request(stock, pos, 'take_profit')
self.log.debug(f"股票 {stock} 实现100%盈利，执行止盈卖出。")
elif pos.price < pos.avg_cost * self.stoploss_stock_rate_data:
# 通过持仓监控器注册卖出请求，而不是直接卖出
self.strategy.state.set_sell_request(stock, pos, 'stop_loss')
self.log.debug(f"股票 {stock} 触及止损阈值，执行卖出。")
self.strategy.state.set_sell_reason("stoploss")
if self.stoploss_strategy == 2 or self.stoploss_strategy == 3:
# 大盘止损判断，若整体市场跌幅过大则平仓所有股票
stock_list = DataHelper.get_index_stocks(self.index_stocks)
df = DataHelper.get_price_safe(
stock_list,
end_date=context.previous_date,
frequency='daily',
fields=['close', 'open'],
count=1,
panel=False
)
if df is not None and not df.empty:
down_ratio = (df['close'] / df['open']).mean()
if down_ratio <= self.stoploss_market_rate_data:
self.strategy.state.set_sell_reason("stoploss")
self.log.debug(f"市场检测到跌幅（平均跌幅 {down_ratio:.2%}），卖出所有持仓。")
for stock in list(context.portfolio.positions.keys()):
pos = context.portfolio.positions[stock]
# 通过持仓监控器注册卖出请求，而不是直接卖出
self.position_monitor.register_sell_request(stock, pos, 'market_stop_loss')

58
task/check_remain_amount_func_task.py Normal file
View File

@ -0,0 +1,58 @@
from typing import Any
from base.base_task import base_task
from kit.kit import DataHelper
from task.process_pending_buys_func_task import process_pending_buys_func_task
class check_remain_amount_func_task(base_task):
"""
检查账户资金与持仓数量：
如果因涨停破板卖出导致持仓不足，则从目标股票中筛选未买入股票，进行补仓操作。
参数:
context: 聚宽平台传入的交易上下文对象
"""
def __init__(self, strategy: Any, sub_task=None):
super().__init__(strategy, sub_task)
def init(self, context: Any):
self.name = "check_remain_amount_func_task"
self.remark = "检查剩余可用资金任务"
self.memo = "检查是否有剩余可用资金进行补仓"
def begin(self, context: Any):
self.stock_num = self.strategy_config.config['stock_num']
self.reason_to_sell = self.strategy.get_reason_to_sell()
self.target_list = self.strategy.get_target_list()
self.not_buy_again = self.position_manager.get_not_buy_again()
def run(self, context: Any):
if self.reason_to_sell[0]:
# 更新持仓列表
self.hold_list = [position.security for position in list(context.portfolio.positions.values())]
if len(self.hold_list) < self.stock_num:
# 处理卖出后的补仓需求
target_list = DataHelper.filter_not_buy_again(self.target_list, self.not_buy_again)
if target_list:
# 将候选股票加入待买入队列
self.position_manager.register_buy_request(target_list[:self.stock_num - len(self.hold_list)])
self.log.info(
f"检测到补仓需求，可用资金 {round(context.portfolio.cash, 2)}，添加 {len(target_list[:self.stock_num - len(self.hold_list)])} 只股票到待买入队列")
# 立即处理待买入请求
task = process_pending_buys_func_task(self.strategy, self.name)
task.process(context)
# 重置卖出原因
self.reason_to_sell = ['']
else:
self.log.info("未检测到涨停破板或止损卖出事件，不进行补仓买入。")
def handle(self, context: Any):
pass
def end(self, context: Any):
pass

101
task/check_state_before_task.py Normal file
View File

@ -0,0 +1,101 @@
import json
from typing import Any
from config.network_config import network_config
from kit.kit import DataHelper
from strategy import trade_strategy
from task.base_task import base_task
class check_state_before_task(base_task):
"""
重置state信息
最前置信息
"""
def __init__(self, strategy: trade_strategy, sub_task=None):
super().__init__(strategy, sub_task)
def init(self, context: Any):
self.log.info("--------------------------")
self.log.info("重置运行中环境数据")
def config(self, context: Any):
# 重置state信息
self.hold_stock = [position.security for position in list(context.portfolio.positions.values())]
self.hold_stock_data = {position.security: position for position in list(context.portfolio.positions.values())}
self.yesterday_high_list = []
self.yesterday_low_list = []
def run(self, context: Any):
positions = context.portfolio.positions
for stock in positions:
self.strategy.state.set_hold_stock(stock.security, stock.__dict__)
if self.hold_stock:
# 获取持仓股票昨日数据（包括收盘价、涨停价、跌停价）
df = DataHelper.get_price_safe(
self.hold_stock,
end_date=context.previous_date,
frequency='daily',
fields=['close', 'high_limit', 'low_limit'],
count=1,
panel=False,
fill_paused=False
)
if df is not None and not df.empty:
# 过滤出收盘价等于涨停价的股票，作为昨日涨停股票
self.yesterday_high_list = list(df[df['close'] == df['high_limit']])
self.yesterday_low_list = list(df[df['close'] == df['low_limit']])
def handle(self, context: Any):
for stock in self.yesterday_high_list:
self.strategy.state.set_high_stock(stock['code'], stock)
for stock in self.yesterday_low_list:
self.strategy.state.set_low_stock(stock['code'], stock)
def end(self, context: Any):
self.log.info("-----------------")
self.log.info("一.截止昨日持仓情况:")
self.hold_stock_info()
self.yesterday_high_list_info()
self.yesterday_low_list_info()
self.log.info("二.保存缓存信息")
self.state_save(context)
self.log.info("三.重置缓存信息")
self.strategy.state.reset()
#
def hold_stock_info(self):
if self.hold_stock:
self.log.info(" 持仓信息: {} 只,昨日涨停信息: {} 只,昨日跌停信息: {} 只".format(len(self.hold_stock), len(self.yesterday_high_list),
len(self.yesterday_low_list)))
for positions in self.hold_stock:
data = self.hold_stock_data
self.log.debug(" 持仓信息: {}".format(positions))
else:
self.log.info(" 无持仓")
def yesterday_high_list_info(self):
if self.yesterday_high_list:
self.log.info(" 涨停信息: {} 只".format(len(self.yesterday_high_list)))
for stock in self.yesterday_high_list:
self.log.debug(" 涨停信息: {}".format(stock))
else:
self.log.info(" 无涨停")
def yesterday_low_list_info(self):
if self.yesterday_high_list:
self.log.info(" 跌停信息: {} 只".format(len(self.yesterday_high_list)))
for stock in self.yesterday_high_list:
self.log.debug(" 跌停信息: {}".format(stock))
else:
self.log.info(" 无跌停")
def state_save(self, context: Any):
json_data = json.dumps(self.strategy.state)
trade_date = context.current_dt.date()
network_config.insert_state(self.strategy_config.strategy_id, json_data, trade_date.strftime("%Y-%m-%d %H:%M:%S"))

96
task/process_pending_buy_task.py Normal file
View File

@ -0,0 +1,96 @@
from typing import Any
from kit.kit import DataHelper
from strategy import trade_strategy
from task.base_task import base_task
class process_pending_buy_task(base_task):
"""
处理待买入股票队列
"""
def __init__(self, strategy: trade_strategy):
super().__init__(strategy)
def config(self, context: Any):
self.today_trade_switch = self.get_config('today_trade_switch')
self.stock_num = self.get_config('stock_num')
self.buy_requests_list = self.strategy.state.get_buy_requests()
def run(self, context: Any):
# 如果在特殊月份（1月/4月）或没有待买入股票，直接返回
if self.today_trade_switch or not self.buy_requests_list:
return
self.log.info("开始处理待买入股票队列")
current_data = DataHelper.get_current_data()
# 更新当前持仓列表
self.hold_list = [position.security for position in list(context.portfolio.positions.values())]
position_count = len(self.hold_list)
target_num = int(self.stock_num)
# 如果持仓已满，不再买入
if position_count >= target_num:
self.log.info("当前持仓已满，清空待买入队列")
self.strategy.state.clear_buy_requests()
return
# 计算可买入的股票数量和分配资金
buy_count = min(len(self.buy_requests_list), target_num - position_count)
if buy_count <= 0:
return
# 计算每只股票可分配的资金
try:
value = context.portfolio.cash / buy_count
except ZeroDivisionError:
self.log.error("资金分摊时除零错误")
return
# 确保每只股票有足够的买入金额
if value < 5000: # 假设最小买入金额为5000元
self.log.warning(f"每只股票分配资金不足: {value:.2f}元，取消本次买入")
return
# 逐个买入股票
success_count = 0
for idx, stock in enumerate(self.buy_requests_list[:buy_count]):
# 跳过已持仓的股票
if stock in self.hold_list:
continue
# 检查是否可交易
if current_data[stock].paused:
self.log.warning(f"股票 {stock} 处于停牌状态，跳过买入")
continue
if current_data[stock].high_limit <= current_data[stock].last_price:
self.log.warning(f"股票 {stock} 已涨停，跳过买入")
continue
# 执行买入
if self.strategy.open_position(stock, value):
self.log.info(f"成功买入股票 {stock}，分配资金 {value:.2f}")
# self.position_manager.get_not_buy_again().append(stock)
success_count += 1
# 如果持仓已满，停止买入
if len(context.portfolio.positions) >= target_num:
break
# 清理已处理的股票
# self.position_manager.pending_buys = self.position_manager.pending_buys[buy_count:]
# self.log.info(f"本次共成功买入 {success_count} 只股票，待买入队列还剩 {len(self.position_manager.pending_buys)} 只")
# 输出最新持仓状态
# self.position_manager.log_status(context, [p.security for p in list(context.portfolio.positions.values())])
def handle(self, context: Any):
pass
def end(self, context: Any):
pass

65
task/process_pending_sells_task.py Normal file
View File

@ -0,0 +1,65 @@
from typing import Any
from kit.kit import DataHelper
from strategy import trade_strategy
from task.base_task import base_task
class process_pending_sells_task(base_task):
"""
处理待卖出股票队列
"""
def __init__(self, strategy: trade_strategy):
super().__init__(strategy)
def config(self, context: Any):
self.sell_request_list = self.strategy.state.get_sell_requests()
def run(self, context: Any):
# 如果没有待卖出股票，直接返回
if len(self.sell_request_list) == 0:
return
self.log.info("开始处理待卖出股票队列")
current_data = DataHelper.get_current_data()
for stock in self.sell_request_list:
if stock not in context.portfolio.positions:
# 股票已不在持仓中，可能已通过其他方式卖出
continue
position = context.portfolio.positions[stock]
if position.closeable_amount <= 0:
# 没有可卖出数量
continue
# 检查是否可以交易
if current_data[stock].paused:
self.log.warning(f"股票 {stock} 已暂停交易，跳过卖出")
continue
if current_data[stock].low_limit >= current_data[stock].last_price:
self.log.warning(f"股票 {stock} 已触及跌停，跳过卖出")
continue
# 尝试卖出
# self.position_monitor.increment_attempt(stock)
success = self.strategy.close_position(context, position)
if success:
# self.position_monitor.mark_as_successful(stock)
self.log.info(f"成功卖出股票 {stock}")
else:
self.log.warning(f"卖出股票 {stock} 失败，已累计尝试 {self.position_monitor.pending_sells[stock]['attempts']} 次")
# 更新日志状态
self.position_monitor.log_status()
def handle(self, context: Any):
pass
def end(self, context: Any):
pass

57
task/trade_afternoon_task.py Normal file
View File

@ -0,0 +1,57 @@
from strategy import trade_strategy
from task.base_task import base_task
from task.check_high_volume_func_task import check_high_volume_func_task
from task.check_limit_up_func_task import check_limit_up_func_task
from task.check_remain_amount_func_task import check_remain_amount_func_task
from task.process_pending_buy_task import process_pending_buy_task
from task.process_pending_sells_task import process_pending_sells_task
class trade_afternoon_task(base_task):
"""
下午交易任务：
1. 检查是否有因为涨停破板触发的卖出信号；
2. 如启用了成交量监控，则检测是否有异常成交量；
3. 检查账户中是否需要补仓；
4. 处理待买入和待卖出队列。
参数:
context: 聚宽平台传入的交易上下文对象
"""
def __init__(self, strategy: trade_strategy):
super().__init__(strategy)
def begin(self, context):
self.no_trading_today_signal = self.config['no_trading_today_signal']
self.HV_control = bool(self.config['HV_control']) # 是否启用成交量监控
def run(self, context):
if not self.no_trading_today_signal:
check_limit_up_task = check_limit_up_func_task(self.strategy, sub_task=self.sub_task)
check_limit_up_task.process(context)
if self.HV_control:
check_high_volume_task = check_high_volume_func_task(self.strategy)
check_high_volume_task.process(context)
# 先处理待卖出股票
process_pending_sells_task = process_pending_sells_task(self.strategy)
process_pending_sells_task.process(context)
# 再检查是否需要补仓
check_remain_amount_task = check_remain_amount_func_task(self.strategy, sub_task=self.sub_task)
check_remain_amount_task.process(context)
# 最后处理待买入股票
sells_task = process_pending_buy_task(self.strategy)
sells_task.process(context)
pass
def handle(self, context):
pass
def end(self, context):
pass

1120
test.py Normal file
View File

File diff suppressed because it is too large Load Diff