| | |
| | | result = strategy_params_settings.settings.to_json_str() |
| | | return {"code": 0, "data": result} |
| | | |
| | | def __on_get_env(self): |
| | | def __on_get_env(self, need_hsitory_data): |
| | | """ |
| | | 获取环境信息 |
| | | @param need_hsitory_data: 是否需要历史数据 |
| | | @return: |
| | | """ |
| | | fdata = {} |
| | |
| | | fdata["real_time_data"] = RealTimeEnvInfo().to_dict() |
| | | # 历史数据 |
| | | fdata["history_data"] = {} |
| | | if tool.get_now_time_str() < '16:00:00': |
| | | # 如果在16:00之前采用当前日期 |
| | | day = tool.get_now_date_str() |
| | | else: |
| | | # 如果在16:00之后采用下一个交易日 |
| | | day = TradeDateManager().get_next_trade_day(tool.get_now_date_str()) |
| | | fdata["history_data"]["leading_limit_up_block_codes_count"] = env_info.get_leading_limit_up_block_codes_count( |
| | | day) |
| | | if need_hsitory_data: |
| | | if tool.get_now_time_str() < '16:00:00': |
| | | # 如果在16:00之前采用当前日期 |
| | | day = tool.get_now_date_str() |
| | | else: |
| | | # 如果在16:00之后采用下一个交易日 |
| | | day = TradeDateManager().get_next_trade_day(tool.get_now_date_str()) |
| | | fdata["history_data"]["leading_limit_up_block_codes_count"] = env_info.get_leading_limit_up_block_codes_count( |
| | | day) |
| | | |
| | | if tool.get_now_time_str() < '16:00:00': |
| | | # 如果在16:00之前采用当前日期 |
| | | day = TradeDateManager().get_previous_trade_day(tool.get_now_date_str()) |
| | | else: |
| | | # 如果在16:00之后采用下一个交易日 |
| | | day = tool.get_now_date_str() |
| | | fdata["history_data"]["k_bars_count"] = env_info.get_history_k_bars(day) |
| | | if tool.get_now_time_str() < '16:00:00': |
| | | # 如果在16:00之前采用当前日期 |
| | | day = TradeDateManager().get_previous_trade_day(tool.get_now_date_str()) |
| | | else: |
| | | # 如果在16:00之后采用下一个交易日 |
| | | day = tool.get_now_date_str() |
| | | fdata["history_data"]["k_bars_count"] = env_info.get_history_k_bars(day) |
| | | return {"code": 0, "data": fdata, "msg": "测试结果"} |
| | | |
| | | def __on_update_leading_limit_up_datas(self): |
| | |
| | | if ctype == "get_settings": |
| | | result_json = self.__on_get_settings() |
| | | elif ctype == 'get_env': |
| | | result_json = self.__on_get_env() |
| | | result_json = self.__on_get_env(data.get("history")) |
| | | elif ctype == 'update_leading_limit_up_datas': |
| | | result_json = self.__on_update_leading_limit_up_datas() |
| | | self.send_response(result_json, client_id, request_id) |