import core.logger as logging from datetime import datetime from time import sleep import pandas as pd import os from core.biz.market_data import MarketData from core.trade.ma_break_statistics import MaBreakStatistics from core.db.db_market_data import DBMarketData from core.biz.metrics_calculation import MetricsCalculation from core.utils import ( datetime_to_timestamp, timestamp_to_datetime, transform_date_time_to_timestamp, ) from trade_data_main import TradeDataMain from config import ( API_KEY, SECRET_KEY, PASSPHRASE, SANDBOX, OKX_MONITOR_CONFIG, BAR_THRESHOLD, ) logger = logging.logger class TradeMaStrategyMain: def __init__( self, is_us_stock: bool = False, is_astock: bool = False, is_aindex: bool = True, is_binance: bool = False, commission_per_share: float = 0, buy_by_long_period: dict = {"by_week": False, "by_month": False}, long_period_condition: dict = { "ma5>ma10": False, "ma10>ma20": False, "macd_diff>0": False, "macd>0": False, }, cut_loss_by_valleys_median: bool = False, ): self.ma_break_statistics = MaBreakStatistics( is_us_stock=is_us_stock, is_astock=is_astock, is_aindex=is_aindex, is_binance=is_binance, commission_per_share=commission_per_share, buy_by_long_period=buy_by_long_period, long_period_condition=long_period_condition, cut_loss_by_valleys_median=cut_loss_by_valleys_median, ) def batch_ma_break_statistics(self): """ 批量计算MA突破统计 """ logger.info("开始批量计算MA突破统计") strategy_dict = self.ma_break_statistics.main_strategy account_value_chg_df_list = [] for strategy_name, strategy_info in strategy_dict.items(): if "macd" in strategy_name: # 只计算macd策略 account_value_chg_df = self.ma_break_statistics.batch_statistics( strategy_name=strategy_name ) account_value_chg_df_list.append(account_value_chg_df) total_account_value_chg_df = pd.concat(account_value_chg_df_list) return total_account_value_chg_df def statistics_account_value_chg(self, account_value_chg_df: pd.DataFrame): logger.info("开始统计account_value_chg") def test_single_symbol(): ma_break_statistics = MaBreakStatistics( is_us_stock=False, is_astock=True, is_aindex=False, is_binance=False, commission_per_share=0, ) symbol = "600111.SH" bar = "1D" ma_break_statistics.trade_simulate( symbol=symbol, bar=bar, strategy_name="均线macd结合策略2" ) def batch_run_strategy(): commission_per_share_list = [0, 0.0008] # cut_loss_by_valleys_median_list = [True, False] cut_loss_by_valleys_median_list = [False] buy_by_long_period_list = [ # {"by_week": True, "by_month": True, "buy_by_10_percentile": True}, {"by_week": False, "by_month": True, "buy_by_10_percentile": True}, {"by_week": True, "by_month": False, "buy_by_10_percentile": True}, {"by_week": False, "by_month": False, "buy_by_10_percentile": False}, ] # buy_by_long_period_list = [{"by_week": False, "by_month": False}] long_period_condition_list = [ {"ma5>ma10": True, "ma10>ma20": True, "macd_diff>0": True, "macd>0": True}, {"ma5>ma10": True, "ma10>ma20": False, "macd_diff>0": True, "macd>0": True}, {"ma5>ma10": False, "ma10>ma20": True, "macd_diff>0": True, "macd>0": True}, ] for commission_per_share in commission_per_share_list: for cut_loss_by_valleys_median in cut_loss_by_valleys_median_list: for buy_by_long_period in buy_by_long_period_list: for long_period_condition in long_period_condition_list: logger.info( f"开始计算, 主要参数：commission_per_share: {commission_per_share}, buy_by_long_period: {buy_by_long_period}, long_period_condition: {long_period_condition}" ) trade_ma_strategy_main = TradeMaStrategyMain( is_us_stock=False, is_astock=False, is_aindex=True, is_binance=False, commission_per_share=commission_per_share, buy_by_long_period=buy_by_long_period, long_period_condition=long_period_condition, cut_loss_by_valleys_median=cut_loss_by_valleys_median, ) trade_ma_strategy_main.batch_ma_break_statistics() trade_ma_strategy_main = TradeMaStrategyMain( is_us_stock=False, is_astock=True, is_aindex=False, is_binance=False, commission_per_share=commission_per_share, buy_by_long_period=buy_by_long_period, long_period_condition=long_period_condition, cut_loss_by_valleys_median=cut_loss_by_valleys_median, ) trade_ma_strategy_main.batch_ma_break_statistics() def pickup_data_from_excel(): main_path = r"./output/trade_sandbox/ma_strategy" sub_main_paths = ["aindex", "astock"] fix_sub_path = "no_cut_loss_by_valleys_median" sub_folder = r"excel/均线macd结合策略2/" file_feature_name = "with_commission" original_df_columns = [ "strategy_name", "symbol", "symbol_name", "bar", "total_buy_commission", "total_sell_commission", "total_commission", "initial_account_value", "final_account_value", "account_value_chg", "market_pct_chg", ] original_df_list = [] for sub_main_path in sub_main_paths: logger.info(f"开始读取{sub_main_path}数据") full_sub_main_path = os.path.join(main_path, sub_main_path, fix_sub_path) # 读取sub_main_path下的所有文件夹 folder_list = os.listdir(full_sub_main_path) for folder in folder_list: logger.info(f"开始读取{folder}数据") folder_path = os.path.join(full_sub_main_path, folder) properties = get_properties_by_folder_name(folder, sub_main_path) logger.info(f"开始读取{folder}数据") # 读取folder_path的sub_folder下的所有文件 sub_folder_path = os.path.join(folder_path, sub_folder) file_list = os.listdir(sub_folder_path) for file in file_list: logger.info(f"开始读取{file}数据") if file_feature_name in file: file_path = os.path.join(sub_folder_path, file) df = pd.read_excel(file_path, sheet_name="资产价值变化") logger.info(f"开始读取{file}数据") # 向df添加properties df = df.assign(**properties) df = df[list(properties.keys()) + original_df_columns] # 将df添加到original_df_list original_df_list.append(df) final_df = pd.concat(original_df_list) excel_folder_path = os.path.join(main_path, "aindex_astock_均线macd结合策略2") os.makedirs(excel_folder_path, exist_ok=True) excel_file_path = os.path.join( excel_folder_path, "all_strategy_with_commission.xlsx" ) with pd.ExcelWriter(excel_file_path) as writer: final_df.to_excel( writer, sheet_name="all_strategy_with_commission", index=False ) def get_properties_by_folder_name(folder_name: str, symbol_type: str): properties = {} sub_properties = folder_name.split("_") properties["symbol_type"] = symbol_type properties["buy_by_long_period"] = "no_long_period" properties["long_period_ma5gtma10"] = False properties["long_period_ma10gtma20"] = False properties["long_period_macd_diffgt0"] = False properties["long_period_macdgt0"] = False properties["buy_by_long_period_10_percentile"] = False if "1M" in sub_properties: properties["buy_by_long_period"] = "1M" if "1W" in sub_properties: properties["buy_by_long_period"] = "1W" if "1W1M" in sub_properties: properties["buy_by_long_period"] = "1W1M" if "ma5gtma10" in sub_properties: properties["long_period_ma5gtma10"] = True if "ma10gtma20" in sub_properties: properties["long_period_ma10gtma20"] = True if "macd_diffgt0" in sub_properties: properties["long_period_macd_diffgt0"] = True if "macdgt0" in sub_properties: properties["long_period_macdgt0"] = True if "10percentile" in sub_properties: properties["buy_by_long_period_10_percentile"] = True return properties def profit_loss_ratio(): """ 计算利润损失比 公式：盈利/盈利交易次数 : 亏损/亏损交易次数 """ folder = r"./output/trade_sandbox/ma_strategy/binance/excel/均线macd结合策略2/" prefix = ["无交易费用", "有交易费用"] for prefix in prefix: excel_file_path = os.path.join( folder, f"{prefix}_趋势投资_from_201708181600_to_202509020600.xlsx" ) df = pd.read_excel(excel_file_path, sheet_name="买卖记录明细") symbol_list = list(df["symbol"].unique()) bar_list = list(df["bar"].unique()) data_list = [] for symbol in symbol_list: for bar in bar_list: df_symbol_bar = df[df["symbol"] == symbol][df["bar"] == bar] start_date = df_symbol_bar["begin_date_time"].min() end_date = df_symbol_bar["end_date_time"].max() profit_df = df_symbol_bar[df_symbol_bar["profit_loss"] > 0] loss_df = df_symbol_bar[df_symbol_bar["profit_loss"] < 0] profit_amount = sum(profit_df["profit_loss"]) loss_amount = abs(sum(loss_df["profit_loss"])) profit_count = len(profit_df) loss_count = len(loss_df) if profit_count == 0 or loss_count == 0: continue profit_loss_ratio = round( (profit_amount / profit_count) / (loss_amount / loss_count) * 100, 4 ) data_list.append( { "币种": symbol, "交易周期": bar, "开始时间": start_date, "结束时间": end_date, "盈利金额": profit_amount, "盈利次数": profit_count, "亏损金额": loss_amount, "亏损次数": loss_count, "盈亏比": profit_loss_ratio, "盈亏比公式": f"盈利金额/盈利次数 : 亏损金额/亏损次数", } ) final_df = pd.DataFrame(data_list) final_df.to_excel(os.path.join(folder, f"{prefix}时虚拟货币利润损失比.xlsx"), index=False) if __name__ == "__main__": # batch_run_strategy() profit_loss_ratio()