import pandas as pd import numpy as np from datetime import datetime import os import warnings warnings.filterwarnings('ignore') class ForexDataProcessor: def __init__(self): self.pairs = ["EURUSD", "GBPUSD", "USDJPY", "USDCHF"] self.data = {} self.processed_data = {} def load_data(self): """Load data for all currency pairs""" success = True for pair in self.pairs: filename = f"{pair}_H1.csv" try: df = pd.read_csv(filename, encoding='utf-16', sep='\t', names=['DateTime', 'Open', 'High', 'Low', 'Close', 'Volume']) # Удаляем строки с дубликатами заголовков df = df[df['DateTime'] != 'DateTime'] # Преобразуем типы данных df['DateTime'] = pd.to_datetime(df['DateTime'], format='%Y.%m.%d %H:%M') for col in ['Open', 'High', 'Low', 'Close']: df[col] = pd.to_numeric(df[col], errors='coerce') df['Volume'] = pd.to_numeric(df['Volume'], errors='coerce') # Удаляем строки с NaN df = df.dropna() df.set_index('DateTime', inplace=True) self.data[pair] = df print(f"Loaded {pair} data successfully. Shape: {df.shape}") except Exception as e: print(f"Error loading {pair} data: {str(e)}") success = False return success def safe_qcut(self, series, q, labels): """Безопасное квантование с обработкой ошибок""" try: if series.nunique() <= q: # Если уникальных значений меньше чем квантилей, используем обычную категоризацию return pd.qcut(series, q=q, labels=labels, duplicates='drop') return pd.qcut(series, q=q, labels=labels) except Exception as e: print(f"Warning: Error in qcut - {str(e)}. Using manual categorization.") # Ручная категоризация как запасной вариант percentiles = np.percentile(series, [20, 40, 60, 80]) return pd.cut(series, bins=[-np.inf] + list(percentiles) + [np.inf], labels=labels) def calculate_indicators(self, df): """Calculate technical indicators for a single dataframe""" result = df.copy() # Базовые расчеты result['Returns'] = result['Close'].pct_change() result['Log_Returns'] = np.log(result['Close']/result['Close'].shift(1)) result['Range'] = result['High'] - result['Low'] result['Range_Pct'] = result['Range'] / result['Open'] * 100 # SMA расчеты for period in [5, 10, 20, 50, 200]: result[f'SMA_{period}'] = result['Close'].rolling(window=period).mean() # EMA расчеты for period in [5, 10, 20, 50]: result[f'EMA_{period}'] = result['Close'].ewm(span=period, adjust=False).mean() # Волатильность result['Volatility'] = result['Returns'].rolling(window=20).std() * np.sqrt(20) # RSI delta = result['Close'].diff() gain = (delta.where(delta > 0, 0)).rolling(window=14).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean() rs = gain / loss result['RSI'] = 100 - (100 / (1 + rs)) # MACD exp1 = result['Close'].ewm(span=12, adjust=False).mean() exp2 = result['Close'].ewm(span=26, adjust=False).mean() result['MACD'] = exp1 - exp2 result['MACD_Signal'] = result['MACD'].ewm(span=9, adjust=False).mean() result['MACD_Hist'] = result['MACD'] - result['MACD_Signal'] # Bollinger Bands result['BB_Middle'] = result['Close'].rolling(window=20).mean() result['BB_Upper'] = result['BB_Middle'] + (result['Close'].rolling(window=20).std() * 2) result['BB_Lower'] = result['BB_Middle'] - (result['Close'].rolling(window=20).std() * 2) result['BB_Width'] = (result['BB_Upper'] - result['BB_Lower']) / result['BB_Middle'] # Дискретизация для ассоциативных правил # Тренд на основе SMA result['Trend'] = 'Sideways' result.loc[result['Close'] > result['SMA_50'], 'Trend'] = 'Uptrend' result.loc[result['Close'] < result['SMA_50'], 'Trend'] = 'Downtrend' # RSI зоны result['RSI_Zone'] = pd.cut(result['RSI'].fillna(50), bins=[-np.inf, 30, 45, 55, 70, np.inf], labels=['Oversold', 'Weak', 'Neutral', 'Strong', 'Overbought']) # Безопасное квантование для остальных показателей labels = ['Very_Low', 'Low', 'Medium', 'High', 'Very_High'] result['Volatility_Zone'] = self.safe_qcut( result['Volatility'].fillna(result['Volatility'].mean()), 5, labels) result['Price_Zone'] = self.safe_qcut( result['Close'], 5, labels) result['Volume_Zone'] = self.safe_qcut( result['Volume'], 5, labels) # Паттерны свечей result['Body'] = result['Close'] - result['Open'] result['Upper_Shadow'] = result['High'] - result[['Open', 'Close']].max(axis=1) result['Lower_Shadow'] = result[['Open', 'Close']].min(axis=1) - result['Low'] result['Body_Pct'] = result['Body'] / result['Open'] * 100 body_mean = abs(result['Body_Pct']).mean() result['Candle_Pattern'] = 'Normal' result.loc[abs(result['Body_Pct']) < body_mean * 0.1, 'Candle_Pattern'] = 'Doji' result.loc[result['Body_Pct'] > body_mean * 2, 'Candle_Pattern'] = 'Long_Bullish' result.loc[result['Body_Pct'] < -body_mean * 2, 'Candle_Pattern'] = 'Long_Bearish' return result def process_all_pairs(self): """Process all currency pairs and create combined dataset""" if not self.load_data(): return None # Обработка каждой пары for pair in self.pairs: if not self.data[pair].empty: print(f"Processing {pair}...") self.processed_data[pair] = self.calculate_indicators(self.data[pair]) # Добавляем префикс пары к названиям колонок self.processed_data[pair].columns = [f"{pair}_{col}" for col in self.processed_data[pair].columns] else: print(f"Skipping {pair} - no data") # Находим общий временной диапазон для непустых данных common_dates = None for pair in self.pairs: if pair in self.processed_data and not self.processed_data[pair].empty: if common_dates is None: common_dates = set(self.processed_data[pair].index) else: common_dates &= set(self.processed_data[pair].index) if not common_dates: print("No common dates found") return None # Выравниваем все пары по общим датам aligned_data = {} for pair in self.pairs: if pair in self.processed_data and not self.processed_data[pair].empty: aligned_data[pair] = self.processed_data[pair].loc[sorted(common_dates)] # Объединяем все пары combined_df = pd.concat([aligned_data[pair] for pair in aligned_data], axis=1) return combined_df def save_data(self, data, suffix='combined'): """Save processed data to CSV""" timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') filename = f"forex_data_{suffix}_{timestamp}.csv" try: data.to_csv(filename, sep='\t', encoding='utf-16') print(f"Saved processed data to: {filename}") return True except Exception as e: print(f"Error saving data: {str(e)}") return False if __name__ == "__main__": processor = ForexDataProcessor() # Обработка всех пар combined_data = processor.process_all_pairs() if combined_data is not None: # Сохраняем объединенный датасет processor.save_data(combined_data) # Выводим информацию о датасете print("\nCombined dataset shape:", combined_data.shape) print("\nFeatures for association rules analysis:") for col in combined_data.columns: if any(x in col for x in ['_Zone', '_Pattern', 'Trend']): print(f"- {col}") # Сохраняем отдельные пары for pair in processor.pairs: if pair in processor.processed_data and not processor.processed_data[pair].empty: processor.save_data(processor.processed_data[pair], pair) import pandas as pd import numpy as np from collections import defaultdict from itertools import combinations import time import logging # Настройка логирования logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('apriori_forex_advanced.log'), logging.StreamHandler() ] ) class AdvancedForexApriori: def __init__(self, min_support=0.01, min_confidence=0.7, max_length=3): self.min_support = min_support self.min_confidence = min_confidence self.max_length = max_length def find_patterns(self, df): start_time = time.time() logging.info("Starting advanced pattern search...") # Группируем колонки по типам для более осмысленного анализа column_groups = { 'trend': [col for col in df.columns if 'Trend' in col], 'rsi': [col for col in df.columns if 'RSI_Zone' in col], 'volume': [col for col in df.columns if 'Volume_Zone' in col], 'price': [col for col in df.columns if 'Price_Zone' in col], 'pattern': [col for col in df.columns if 'Pattern' in col] } # Создаем список всех колонок для анализа pattern_cols = [] for cols in column_groups.values(): pattern_cols.extend(cols) logging.info(f"Found {len(pattern_cols)} pattern columns in {len(column_groups)} groups") # Готовим данные pattern_df = df[pattern_cols] n_rows = len(pattern_df) # Находим одиночные паттерны logging.info("Finding single patterns...") single_patterns = {} for col in pattern_cols: value_counts = pattern_df[col].value_counts() value_counts = value_counts[value_counts/n_rows >= self.min_support] for value, count in value_counts.items(): pattern = f"{col}={value}" single_patterns[pattern] = count/n_rows # Находим парные и тройные паттерны logging.info("Finding complex patterns...") complex_rules = [] # Генерируем комбинации колонок для анализа column_combinations = [] for i in range(2, self.max_length + 1): column_combinations.extend(combinations(pattern_cols, i)) total_combinations = len(column_combinations) for idx, cols in enumerate(column_combinations, 1): if idx % 10 == 0: logging.info(f"Processing combination {idx}/{total_combinations}") # Создаем кросс-таблицу для выбранных колонок grouped = pattern_df.groupby([*cols]).size().reset_index(name='count') grouped['support'] = grouped['count'] / n_rows # Фильтруем по минимальной поддержке grouped = grouped[grouped['support'] >= self.min_support] for _, row in grouped.iterrows(): # Формируем все возможные комбинации антецедентов и консеквентов items = [f"{col}={row[col]}" for col in cols] for i in range(1, len(items)): for antecedent in combinations(items, i): consequent = tuple(set(items) - set(antecedent)) # Считаем поддержку антецедента ant_support = self._calculate_support(pattern_df, antecedent) if ant_support > 0: # Избегаем деления на ноль confidence = row['support'] / ant_support if confidence >= self.min_confidence: # Считаем лифт cons_support = self._calculate_support(pattern_df, consequent) lift = confidence / cons_support if cons_support > 0 else 0 # Добавляем дополнительные метрики для оценки правил leverage = row['support'] - (ant_support * cons_support) conviction = (1 - cons_support) / (1 - confidence) if confidence < 1 else float('inf') rule = { 'antecedent': antecedent, 'consequent': consequent, 'support': row['support'], 'confidence': confidence, 'lift': lift, 'leverage': leverage, 'conviction': conviction } # Фильтруем правила по дополнительным критериям if self._is_meaningful_rule(rule): complex_rules.append(rule) # Сортируем правила по комплексной метрике complex_rules.sort(key=self._rule_score, reverse=True) end_time = time.time() logging.info(f"Pattern search completed in {end_time - start_time:.2f} seconds") logging.info(f"Found {len(complex_rules)} meaningful rules") return complex_rules def _calculate_support(self, df, items): """Вычисляет поддержку для набора элементов""" mask = pd.Series(True, index=df.index) for item in items: col, val = item.split('=') mask &= (df[col] == val) return mask.mean() def _is_meaningful_rule(self, rule): """Проверяет правило на значимость для трейдинга""" # Правило должно иметь высокий лифт и leverage if rule['lift'] < 1.5 or rule['leverage'] < 0.01: return False # Хотя бы один элемент должен быть связан с трендом или RSI has_trend_or_rsi = any('Trend' in item or 'RSI' in item for item in rule['antecedent'] + rule['consequent']) if not has_trend_or_rsi: return False return True def _rule_score(self, rule): """Вычисляет комплексную оценку правила""" return (rule['lift'] * 0.4 + rule['confidence'] * 0.3 + rule['support'] * 0.2 + rule['leverage'] * 0.1) # Загрузка данных logging.info("Loading data...") data = pd.read_csv('forex_data_combined_20241116_074242.csv', sep='\t', encoding='utf-16', index_col='DateTime') logging.info(f"Data loaded, shape: {data.shape}") # Применение алгоритма apriori = AdvancedForexApriori(min_support=0.01, min_confidence=0.7, max_length=3) rules = apriori.find_patterns(data) # Вывод результатов logging.info("\nTop 10 trading rules:") for i, rule in enumerate(rules[:10], 1): logging.info(f"\nRule {i}:") logging.info(f"IF {' AND '.join(rule['antecedent'])}") logging.info(f"THEN {' AND '.join(rule['consequent'])}") logging.info(f"Support: {rule['support']:.3f}") logging.info(f"Confidence: {rule['confidence']:.3f}") logging.info(f"Lift: {rule['lift']:.3f}") logging.info(f"Leverage: {rule['leverage']:.3f}") logging.info(f"Conviction: {rule['conviction']:.3f}") # Сохранение результатов results_df = pd.DataFrame(rules) results_df.to_csv('forex_rules_advanced.csv', index=False, sep='\t', encoding='utf-16') logging.info("Results saved to forex_rules_advanced.csv") import pandas as pd import numpy as np from collections import defaultdict from itertools import combinations import time import logging # Настройка логирования logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('apriori_forex_advanced.log'), logging.StreamHandler() ] ) class AdvancedForexApriori: def __init__(self, min_support=0.01, min_confidence=0.7, max_length=3): self.min_support = min_support self.min_confidence = min_confidence self.max_length = max_length def find_patterns(self, df): start_time = time.time() logging.info("Starting advanced pattern search...") # Группируем колонки по типам для более осмысленного анализа column_groups = { 'trend': [col for col in df.columns if 'Trend' in col], 'rsi': [col for col in df.columns if 'RSI_Zone' in col], 'volume': [col for col in df.columns if 'Volume_Zone' in col], 'price': [col for col in df.columns if 'Price_Zone' in col], 'pattern': [col for col in df.columns if 'Pattern' in col] } # Создаем список всех колонок для анализа pattern_cols = [] for cols in column_groups.values(): pattern_cols.extend(cols) logging.info(f"Found {len(pattern_cols)} pattern columns in {len(column_groups)} groups") # Готовим данные pattern_df = df[pattern_cols] n_rows = len(pattern_df) # Находим одиночные паттерны logging.info("Finding single patterns...") single_patterns = {} for col in pattern_cols: value_counts = pattern_df[col].value_counts() value_counts = value_counts[value_counts/n_rows >= self.min_support] for value, count in value_counts.items(): pattern = f"{col}={value}" single_patterns[pattern] = count/n_rows # Находим парные и тройные паттерны logging.info("Finding complex patterns...") complex_rules = [] # Генерируем комбинации колонок для анализа column_combinations = [] for i in range(2, self.max_length + 1): column_combinations.extend(combinations(pattern_cols, i)) total_combinations = len(column_combinations) for idx, cols in enumerate(column_combinations, 1): if idx % 10 == 0: logging.info(f"Processing combination {idx}/{total_combinations}") # Создаем кросс-таблицу для выбранных колонок grouped = pattern_df.groupby([*cols]).size().reset_index(name='count') grouped['support'] = grouped['count'] / n_rows # Фильтруем по минимальной поддержке grouped = grouped[grouped['support'] >= self.min_support] for _, row in grouped.iterrows(): # Формируем все возможные комбинации антецедентов и консеквентов items = [f"{col}={row[col]}" for col in cols] for i in range(1, len(items)): for antecedent in combinations(items, i): consequent = tuple(set(items) - set(antecedent)) # Считаем поддержку антецедента ant_support = self._calculate_support(pattern_df, antecedent) if ant_support > 0: # Избегаем деления на ноль confidence = row['support'] / ant_support if confidence >= self.min_confidence: # Считаем лифт cons_support = self._calculate_support(pattern_df, consequent) lift = confidence / cons_support if cons_support > 0 else 0 # Добавляем дополнительные метрики для оценки правил leverage = row['support'] - (ant_support * cons_support) conviction = (1 - cons_support) / (1 - confidence) if confidence < 1 else float('inf') rule = { 'antecedent': antecedent, 'consequent': consequent, 'support': row['support'], 'confidence': confidence, 'lift': lift, 'leverage': leverage, 'conviction': conviction } # Фильтруем правила по дополнительным критериям if self._is_meaningful_rule(rule): complex_rules.append(rule) # Сортируем правила по комплексной метрике complex_rules.sort(key=self._rule_score, reverse=True) end_time = time.time() logging.info(f"Pattern search completed in {end_time - start_time:.2f} seconds") logging.info(f"Found {len(complex_rules)} meaningful rules") return complex_rules def _calculate_support(self, df, items): """Вычисляет поддержку для набора элементов""" mask = pd.Series(True, index=df.index) for item in items: col, val = item.split('=') mask &= (df[col] == val) return mask.mean() def _is_meaningful_rule(self, rule): """Проверяет правило на значимость для трейдинга""" # Правило должно иметь высокий лифт и leverage if rule['lift'] < 1.5 or rule['leverage'] < 0.01: return False # Хотя бы один элемент должен быть связан с трендом или RSI has_trend_or_rsi = any('Trend' in item or 'RSI' in item for item in rule['antecedent'] + rule['consequent']) if not has_trend_or_rsi: return False return True def _rule_score(self, rule): """Вычисляет комплексную оценку правила""" return (rule['lift'] * 0.4 + rule['confidence'] * 0.3 + rule['support'] * 0.2 + rule['leverage'] * 0.1) # Загрузка данных logging.info("Loading data...") data = pd.read_csv('forex_data_combined_20241116_074242.csv', sep='\t', encoding='utf-16', index_col='DateTime') logging.info(f"Data loaded, shape: {data.shape}") # Применение алгоритма apriori = AdvancedForexApriori(min_support=0.01, min_confidence=0.7, max_length=3) rules = apriori.find_patterns(data) # Вывод результатов logging.info("\nTop 10 trading rules:") for i, rule in enumerate(rules[:10], 1): logging.info(f"\nRule {i}:") logging.info(f"IF {' AND '.join(rule['antecedent'])}") logging.info(f"THEN {' AND '.join(rule['consequent'])}") logging.info(f"Support: {rule['support']:.3f}") logging.info(f"Confidence: {rule['confidence']:.3f}") logging.info(f"Lift: {rule['lift']:.3f}") logging.info(f"Leverage: {rule['leverage']:.3f}") logging.info(f"Conviction: {rule['conviction']:.3f}") # Сохранение результатов results_df = pd.DataFrame(rules) results_df.to_csv('forex_rules_advanced.csv', index=False, sep='\t', encoding='utf-16') logging.info("Results saved to forex_rules_advanced.csv") import pandas as pd import numpy as np import matplotlib.pyplot as plt from pyvis.network import Network import plotly.graph_objects as go import plotly.express as px class ForexRulesVisualizer: def __init__(self, rules_df): self.rules_df = rules_df def plot_metrics_distribution(self): """Визуализация распределения метрик правил""" fig, axes = plt.subplots(2, 2, figsize=(15, 10)) # Support distribution axes[0,0].hist(self.rules_df['support'], bins=30) axes[0,0].set_title('Distribution of Support') axes[0,0].set_xlabel('Support') axes[0,0].set_ylabel('Frequency') # Confidence distribution axes[0,1].hist(self.rules_df['confidence'], bins=30) axes[0,1].set_title('Distribution of Confidence') axes[0,1].set_xlabel('Confidence') axes[0,1].set_ylabel('Frequency') # Lift distribution axes[1,0].hist(self.rules_df['lift'], bins=30) axes[1,0].set_title('Distribution of Lift') axes[1,0].set_xlabel('Lift') axes[1,0].set_ylabel('Frequency') # Leverage distribution axes[1,1].hist(self.rules_df['leverage'], bins=30) axes[1,1].set_title('Distribution of Leverage') axes[1,1].set_xlabel('Leverage') axes[1,1].set_ylabel('Frequency') plt.tight_layout() return fig def create_network_graph(self, min_lift=1.5, min_confidence=0.7): """Создает интерактивный граф связей между правилами""" net = Network(height='750px', width='100%', bgcolor='#ffffff', font_color='black') # Фильтруем правила filtered_rules = self.rules_df[ (self.rules_df['lift'] >= min_lift) & (self.rules_df['confidence'] >= min_confidence) ] # Добавляем узлы и ребра for _, rule in filtered_rules.iterrows(): # Антецеденты for item in rule['antecedent']: net.add_node(item, title=item, color='#97c2fc') # Консеквенты for item in rule['consequent']: net.add_node(item, title=item, color='#fb7e81') # Связи между ними for ant in rule['antecedent']: for cons in rule['consequent']: net.add_edge(ant, cons, value=rule['lift'], title=f"Lift: {rule['lift']:.2f}\n" \ f"Conf: {rule['confidence']:.2f}") return net def plot_heatmap_matrix(self): """Создает тепловую карту связей между парами валют""" pairs = ['EURUSD', 'GBPUSD', 'USDJPY', 'USDCHF'] matrix = pd.DataFrame(0, index=pairs, columns=pairs) for _, rule in self.rules_df.iterrows(): source_pairs = set() target_pairs = set() # Находим валютные пары в антецедентах и консеквентах for ant in rule['antecedent']: pair = next((p for p in pairs if p in ant), None) if pair: source_pairs.add(pair) for cons in rule['consequent']: pair = next((p for p in pairs if p in cons), None) if pair: target_pairs.add(pair) # Обновляем матрицу for source in source_pairs: for target in target_pairs: matrix.loc[source, target] += rule['lift'] # Нормализуем и создаем тепловую карту matrix = matrix / matrix.max().max() fig = go.Figure(data=go.Heatmap( z=matrix.values, x=matrix.columns, y=matrix.index, colorscale='Viridis', text=np.around(matrix.values, 2), texttemplate='%{text}', textfont={"size": 10}, hoverongaps=False)) fig.update_layout( title='Strength of Relationships Between Currency Pairs', xaxis_title='Target Pair', yaxis_title='Source Pair', width=800, height=600 ) return fig # Загружаем данные с правилами logging.info("Loading generated rules...") results_df = pd.read_csv('forex_rules_advanced.csv', sep='\t', encoding='utf-16') # Исправляем формат данных (строки в списки) results_df['antecedent'] = results_df['antecedent'].apply(eval) results_df['consequent'] = results_df['consequent'].apply(eval) # Применяем визуализацию logging.info("Creating visualizations...") visualizer = ForexRulesVisualizer(results_df) # Создаем и сохраняем визуализации metrics_fig = visualizer.plot_metrics_distribution() metrics_fig.savefig('rule_metrics_distribution.png') logging.info("Metrics distribution plot saved") network_graph = visualizer.create_network_graph() network_graph.save_graph('rule_network.html') logging.info("Network graph saved") heatmap_fig = visualizer.plot_heatmap_matrix() heatmap_fig.write_html('pair_relationships_heatmap.html') logging.info("Heatmap saved") import pandas as pd import numpy as np from datetime import datetime import logging class ForexSignalGenerator: def __init__(self, rules_df, min_rule_strength=0.5): """ Инициализация генератора сигналов Parameters: rules_df: DataFrame с ассоциативными правилами min_rule_strength: минимальная сила правила для генерации сигнала """ self.rules_df = rules_df self.min_rule_strength = min_rule_strength self.active_signals = {} def calculate_rule_strength(self, rule): """ Комплексная оценка силы правила Учитывает все метрики с разными весами """ strength = ( rule['lift'] * 0.4 + # Основной вес на lift rule['confidence'] * 0.3 + # Уверенность правила rule['support'] * 0.2 + # Частота появления rule['leverage'] * 0.1 # Улучшение над случайностью ) # Дополнительный бонус за наличие трендовых индикаторов if any('Trend' in item for item in rule['antecedent']): strength *= 1.2 return strength def analyze_market_state(self, current_data): """ Анализ текущего состояния рынка Parameters: current_data: DataFrame с текущими значениями индикаторов """ signals = [] state = self._create_market_state(current_data) # Находим все подходящие правила matching_rules = self._find_matching_rules(state) # Группируем правила по валютным парам for pair in ['EURUSD', 'GBPUSD', 'USDJPY', 'USDCHF']: pair_rules = [r for r in matching_rules if any(pair in c for c in r['consequent'])] if pair_rules: signal = self._generate_pair_signal(pair, pair_rules) signals.append(signal) return signals def _create_market_state(self, data): """Формирование текущего состояния рынка""" state = [] for col in data.columns: if any(x in col for x in ['_Zone', '_Pattern', 'Trend']): state.append(f"{col}={data[col].iloc[-1]}") return set(state) def _find_matching_rules(self, state): """Поиск правил, соответствующих текущему состоянию""" matching_rules = [] for _, rule in self.rules_df.iterrows(): # Проверяем, выполняются ли все условия правила if all(cond in state for cond in rule['antecedent']): strength = self.calculate_rule_strength(rule) if strength >= self.min_rule_strength: rule['calculated_strength'] = strength matching_rules.append(rule) return matching_rules def _generate_pair_signal(self, pair, rules): """Генерация сигнала для конкретной валютной пары""" # Разделяем правила по типу сигнала trend_signals = defaultdict(float) for rule in rules: # Ищем консеквенты, связанные с трендом trend_cons = [c for c in rule['consequent'] if pair in c and 'Trend' in c] if trend_cons: for cons in trend_cons: trend = cons.split('=')[1] trend_signals[trend] += rule['calculated_strength'] # Определяем итоговый сигнал if trend_signals: strongest_trend = max(trend_signals.items(), key=lambda x: x[1]) return { 'pair': pair, 'signal': strongest_trend[0], 'strength': strongest_trend[1], 'timestamp': datetime.now() } return None # Пример использования def run_trading_system(data, rules_df): """ Запуск торговой системы Parameters: data: DataFrame с историческими данными rules_df: DataFrame с ассоциативными правилами """ signal_generator = ForexSignalGenerator(rules_df) # Симулируем проход по историческим данным signals_history = [] for i in range(len(data) - 1): current_slice = data.iloc[i:i+1] signals = signal_generator.analyze_market_state(current_slice) for signal in signals: if signal: signals_history.append({ 'datetime': current_slice.index[0], 'pair': signal['pair'], 'signal': signal['signal'], 'strength': signal['strength'] }) return pd.DataFrame(signals_history) # Загружаем исторические данные и правила data = pd.read_csv('forex_data_combined_20241116_090857.csv', sep='\t', encoding='utf-16', index_col='DateTime', parse_dates=True) rules_df = pd.read_csv('forex_rules_advanced.csv', sep='\t', encoding='utf-16') rules_df['antecedent'] = rules_df['antecedent'].apply(eval) rules_df['consequent'] = rules_df['consequent'].apply(eval) # Запускаем тестирование signals_df = run_trading_system(data, rules_df) # Анализируем результаты print("Generated signals statistics:") print(signals_df.groupby('pair')['signal'].value_counts()) # Запускаем тестирование signals_df = run_trading_system(data, rules_df) # Анализируем результаты print("Generated signals statistics:") print(signals_df.groupby('pair')['signal'].value_counts()) # Печать последних 10 сигналов print("\nLast 10 signals:") print(signals_df.tail(10)[['datetime', 'pair', 'signal', 'strength']])