> Machine Learning Guru|

import yfinance as yf
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import time

class MultiSourceDataCollector:
    def __init__(self, primary_delay=1.0, fallback_delay=2.0):
        self.primary_delay = primary_delay
        self.fallback_delay = fallback_delay
        self.quality_scores = {}
        
    def fetch_with_retry(self, ticker, start_date, end_date, max_retries=3):
        """Получение данных с повторными попытками и обработкой ошибок"""
        for attempt in range(max_retries):
            try:
                time.sleep(self.primary_delay)
                data = yf.download(ticker, start=start_date, end=end_date, 
                                   progress=False, auto_adjust=False)
                
                # Проверка на Multiindex
                if isinstance(data.columns, pd.MultiIndex):
                    data.columns = data.columns.droplevel(1)
                
                if not data.empty:
                    return data, 'yfinance'
            except Exception as e:
                if attempt == max_retries - 1:
                    return pd.DataFrame(), None
                time.sleep(self.fallback_delay * (attempt + 1))
        
        return pd.DataFrame(), None
    
    def validate_data_quality(self, data, ticker):
        """Оценка качества полученных данных"""
        if data.empty:
            return 0.0
        
        quality_score = 100.0
        
        # Проверка на пропуски
        missing_ratio = data['Close'].isna().sum() / len(data)
        quality_score -= missing_ratio * 30
        
        # Проверка на нулевые объемы
        zero_volume_ratio = (data['Volume'] == 0).sum() / len(data)
        quality_score -= zero_volume_ratio * 20
        
        # Проверка на выбросы через IQR
        q1 = data['Close'].quantile(0.25)
        q3 = data['Close'].quantile(0.75)
        iqr = q3 - q1
        outliers = ((data['Close'] < q1 - 3*iqr) | (data['Close'] > q3 + 3*iqr)).sum()
        outlier_ratio = outliers / len(data)
        quality_score -= outlier_ratio * 25
        
        # Проверка монотонности временных меток
        if not data.index.is_monotonic_increasing:
            quality_score -= 15
        
        return max(0.0, quality_score)
    
    def collect_and_merge(self, tickers, start_date, end_date):
        """Сбор данных по множеству тикеров с оценкой качества"""
        results = {}
        
        for ticker in tickers:
            data, source = self.fetch_with_retry(ticker, start_date, end_date)
            
            if not data.empty:
                quality = self.validate_data_quality(data, ticker)
                self.quality_scores[ticker] = {
                    'score': quality,
                    'source': source,
                    'rows': len(data),
                    'missing': data['Close'].isna().sum()
                }
                results[ticker] = data
            else:
                self.quality_scores[ticker] = {
                    'score': 0.0,
                    'source': None,
                    'rows': 0,
                    'missing': 0
                }
        
        return results

# Пример использования
collector = MultiSourceDataCollector(primary_delay=1.2, fallback_delay=2.5)

tickers = ['BABA', 'TSM', 'SHOP', 'HD']
end_date = datetime.now()
start_date = end_date - timedelta(days=365)

data_collection = collector.collect_and_merge(
    tickers, 
    start_date.strftime('%Y-%m-%d'),
    end_date.strftime('%Y-%m-%d')
)

# Анализ качества данных
quality_df = pd.DataFrame(collector.quality_scores).T
print("\nОценка качества данных по тикерам:")
print(quality_df.sort_values('score', ascending=False))

Оценка качества данных по тикерам:
      score    source rows missing
BABA  100.0  yfinance  250       0
TSM   100.0  yfinance  250       0
SHOP  100.0  yfinance  250       0
HD    100.0  yfinance  250       0

import time
import threading
from collections import deque
from datetime import datetime, timedelta
import requests
from typing import Dict, Optional, Callable
import logging
class RateLimiter:
def __init__(self, calls_per_second=5, burst_size=10):
self.calls_per_second = calls_per_second
self.burst_size = burst_size
self.tokens = burst_size
self.last_update = time.time()
self.lock = threading.Lock()
def acquire(self, tokens=1):
"""Алгоритм Token bucket для ограничения скорости"""
with self.lock:
now = time.time()
elapsed = now - self.last_update
# Добавляем токены на основе прошедшего времени
self.tokens = min(
self.burst_size,
self.tokens + elapsed * self.calls_per_second
)
self.last_update = now
if self.tokens >= tokens:
self.tokens -= tokens
return 0  # Нет задержки
else:
# Рассчитываем необходимое время ожидания
wait_time = (tokens - self.tokens) / self.calls_per_second
return wait_time
class AdaptiveAPIClient:
def __init__(self, api_key, base_url, calls_per_minute=5):
self.api_key = api_key
self.base_url = base_url
self.rate_limiter = RateLimiter(
calls_per_second=calls_per_minute/60,
burst_size=calls_per_minute
)
# Статистика для адаптации
self.request_times = deque(maxlen=100)
self.error_count = 0
self.success_count = 0
self.last_429_time = None
# Адаптивные параметры
self.backoff_multiplier = 1.0
self.max_backoff = 5.0
logging.basicConfig(level=logging.INFO)
self.logger = logging.getLogger(__name__)
def _adaptive_wait(self):
"""Адаптивное ожидание на основе истории запросов"""
base_wait = self.rate_limiter.acquire()
# Увеличиваем задержку, если недавно получали 429
if self.last_429_time:
time_since_429 = time.time() - self.last_429_time
if time_since_429 < 300: # 5 минут base_wait *= self.backoff_multiplier if base_wait > 0:
time.sleep(base_wait)
def _handle_rate_limit_error(self, response):
"""Обработка 429 ошибки с экспоненциальным backoff"""
self.last_429_time = time.time()
self.backoff_multiplier = min(
self.backoff_multiplier * 2,
self.max_backoff
)
# Проверяем Retry-After header
retry_after = response.headers.get('Retry-After')
if retry_after:
try:
wait_time = int(retry_after)
self.logger.warning(
f"Rate limit hit. Waiting {wait_time}s as per Retry-After header"
)
time.sleep(wait_time)
return
except ValueError:
pass
# Если нет Retry-After, используем экспоненциальный backoff
wait_time = 60 * self.backoff_multiplier
self.logger.warning(
f"Rate limit hit. Exponential backoff: waiting {wait_time:.1f}s"
)
time.sleep(wait_time)
def request_with_retry(self, endpoint, params=None, max_retries=3):
"""Запрос с автоматическими повторами и адаптацией"""
params = params or {}
params['apikey'] = self.api_key
for attempt in range(max_retries):
try:
self._adaptive_wait()
start_time = time.time()
response = requests.get(
f"{self.base_url}/{endpoint}",
params=params,
timeout=30
)
request_time = time.time() - start_time
self.request_times.append(request_time)
if response.status_code == 200:
self.success_count += 1
# Постепенно уменьшаем backoff при успешных запросах
self.backoff_multiplier = max(
1.0,
self.backoff_multiplier * 0.9
)
return response.json()
elif response.status_code == 429:
self.error_count += 1
self._handle_rate_limit_error(response)
if attempt < max_retries - 1: continue else: raise Exception("Max retries exceeded on rate limit") elif response.status_code >= 500:
# Серверные ошибки - повторяем с задержкой
wait_time = (attempt + 1) ** 2
self.logger.warning(
f"Server error {response.status_code}. "
f"Retry {attempt+1}/{max_retries} after {wait_time}s"
)
time.sleep(wait_time)
continue
else:
self.logger.error(
f"Request failed with status {response.status_code}: "
f"{response.text}"
)
return None
except requests.exceptions.Timeout:
self.logger.warning(f"Request timeout. Attempt {attempt+1}/{max_retries}")
time.sleep((attempt + 1) * 2)
continue
except requests.exceptions.RequestException as e:
self.logger.error(f"Request exception: {e}")
if attempt < max_retries - 1: time.sleep((attempt + 1) * 2) continue return None return None def get_statistics(self): """Статистика использования API""" if not self.request_times: return { 'total_requests': 0, 'successful': 0, 'errors': 0, 'success_rate': 0.0, 'avg_response_time': 0.0, 'current_backoff': self.backoff_multiplier } avg_time = sum(self.request_times) / len(self.request_times) total_requests = self.success_count + self.error_count success_rate = (self.success_count / total_requests * 100) if total_requests > 0 else 0
return {
'total_requests': total_requests,
'successful': self.success_count,
'errors': self.error_count,
'success_rate': success_rate,
'avg_response_time': avg_time,
'current_backoff': self.backoff_multiplier
}
class BatchDataCollector:
def __init__(self, api_client: AdaptiveAPIClient, batch_size=5):
self.api_client = api_client
self.batch_size = batch_size
self.logger = logging.getLogger(__name__)
def collect_multiple_tickers(self, tickers, data_function: Callable):
"""Сбор данных по множеству тикеров с батчингом"""
results = {}
failed = []
total_batches = (len(tickers) + self.batch_size - 1) // self.batch_size
for i in range(0, len(tickers), self.batch_size):
batch = tickers[i:i + self.batch_size]
batch_num = i // self.batch_size + 1
self.logger.info(
f"Processing batch {batch_num}/{total_batches}: {batch}"
)
for ticker in batch:
try:
data = data_function(ticker)
if data is not None:
results[ticker] = data
else:
failed.append(ticker)
except Exception as e:
self.logger.error(f"Error processing {ticker}: {e}")
failed.append(ticker)
# Статистика после каждого батча
stats = self.api_client.get_statistics()
self.logger.info(
f"Batch {batch_num} complete. "
f"Success rate: {stats['success_rate']:.1f}%, "
f"Avg response: {stats['avg_response_time']:.2f}s"
)
# Повторная попытка для failed тикеров
if failed:
self.logger.info(f"Retrying {len(failed)} failed tickers")
time.sleep(30)  # Дополнительная пауза перед retry
for ticker in failed[:]:
try:
data = data_function(ticker)
if data is not None:
results[ticker] = data
failed.remove(ticker)
except Exception as e:
self.logger.error(f"Retry failed for {ticker}: {e}")
return results, failed
# Пример использования
def demo_rate_limiting():
API_KEY = "**************"
client = AdaptiveAPIClient(
api_key=API_KEY,
base_url="https://www.alphavantage.co/query",
calls_per_minute=5  # У бесплатного тарифа лимит 5 запросов/мин
)
collector = BatchDataCollector(client, batch_size=3)
# Список тикеров
tickers = ["BABA", "TSM", "SHOP", "SQ", "AMD", "MU", "INTC", "QCOM"]
def fetch_ticker_data(ticker):
params = {
"function": "TIME_SERIES_DAILY_ADJUSTED",
"symbol": ticker,
"outputsize": "compact"
}
data = client.request_with_retry("", params=params)
# Проверяем, что ответ корректный
if not data:
return None
if "Note" in data:
print(f"⚠️  API limit notice for {ticker}: {data['Note']}")
return None
if "Error Message" in data:
print(f"❌ Error for {ticker}: {data['Error Message']}")
return None
if "Time Series (Daily)" in data:
latest_date, latest_values = list(data["Time Series (Daily)"].items())[0]
return {
"ticker": ticker,
"date": latest_date,
"close": float(latest_values["4. close"]),
"adjusted_close": float(latest_values["5. adjusted close"]),
"volume": int(latest_values["6. volume"])
}
return None
print("Начинаем сбор данных с API Alpha Vantage...")
start_time = time.time()
results, failed = collector.collect_multiple_tickers(
tickers,
fetch_ticker_data
)
elapsed = time.time() - start_time
print(f"\nСбор завершен за {elapsed:.1f} секунд")
print(f"Успешно: {len(results)}, Ошибки: {len(failed)}")
# Вывод результатов
for ticker, info in results.items():
print(f"{ticker}: {info}")
stats = client.get_statistics()
print("\nФинальная статистика:")
for key, value in stats.items():
if isinstance(value, float):
print(f"{key}: {value:.2f}")
else:
print(f"{key}: {value}")
demo_rate_limiting()

Начинаем сбор данных с API Alpha Vantage...
Сбор завершен за 84.0 секунд
Успешно: 8, Ошибки: 0
Финальная статистика:
total_requests: 16
successful: 16
errors: 0
success_rate: 100.00
avg_response_time: 0.71
current_backoff: 1.00

import pandas as pd
import numpy as np
from datetime import datetime
import pytz
from zoneinfo import ZoneInfo
class TimestampNormalizer:
def __init__(self):
self.exchange_timezones = {
'NYSE': 'America/New_York',
'NASDAQ': 'America/New_York',
'LSE': 'Europe/London',
'TSE': 'Asia/Tokyo',
'HKEX': 'Asia/Hong_Kong',
'SSE': 'Asia/Shanghai'
}
self.trading_hours = {
'NYSE': {'open': '09:30', 'close': '16:00'},
'NASDAQ': {'open': '09:30', 'close': '16:00'},
'LSE': {'open': '08:00', 'close': '16:30'},
'TSE': {'open': '09:00', 'close': '15:00'}
}
def normalize_to_utc(self, data, exchange='NYSE'):
"""Конвертация временных меток в UTC"""
if not isinstance(data.index, pd.DatetimeIndex):
data.index = pd.to_datetime(data.index)
tz = self.exchange_timezones.get(exchange, 'America/New_York')
if data.index.tz is not None:
data.index = data.index.tz_convert('UTC')
else:
data.index = data.index.tz_localize(
tz, ambiguous='infer', nonexistent='shift_forward'
)
data.index = data.index.tz_convert('UTC')
return data
def align_trading_days(self, data_dict, method='intersection', agg='first'):
"""
Выравнивание по торговым дням (UTC).
agg: 'first', 'last', 'mean' — как агрегировать внутри дня.
"""
if not data_dict:
return {}
# Переводим все ряды в UTC
normalized = {}
for key, df in data_dict.items():
if not df.empty:
normalized[key] = self.normalize_to_utc(df.copy(), exchange=key)
if not normalized:
return {}
# Приводим к дневным данным
if agg == 'first':
daily = {key: df.groupby(df.index.normalize()).first()
for key, df in normalized.items()}
elif agg == 'last':
daily = {key: df.groupby(df.index.normalize()).last()
for key, df in normalized.items()}
elif agg == 'mean':
daily = {key: df.groupby(df.index.normalize()).mean()
for key, df in normalized.items()}
else:
raise ValueError("agg должен быть 'first', 'last' или 'mean'")
# Совмещение дат
if method == 'intersection':
common_days = daily[list(daily.keys())[0]].index
for df in list(daily.values())[1:]:
common_days = common_days.intersection(df.index)
aligned = {key: df.loc[common_days] for key, df in daily.items()}
elif method == 'union':
all_days = pd.DatetimeIndex([])
for df in daily.values():
all_days = all_days.union(df.index)
aligned = {key: df.reindex(all_days) for key, df in daily.items()}
return aligned
# ДЕМО
normalizer = TimestampNormalizer()
# Создаем тестовые данные (NYSE и LSE)
dates_ny = pd.date_range('2025-01-01', periods=100, freq='D', tz='America/New_York')
dates_london = pd.date_range('2025-01-01', periods=100, freq='D', tz='Europe/London')
test_data_ny = pd.DataFrame({
'Close': np.random.randn(100).cumsum() + 100,
'Volume': np.random.randint(1_000_000, 10_000_000, 100)
}, index=dates_ny)
test_data_london = pd.DataFrame({
'Close': np.random.randn(100).cumsum() + 150,
'Volume': np.random.randint(500_000, 5_000_000, 100)
}, index=dates_london)
# Приводим в UTC
normalized_ny = normalizer.normalize_to_utc(test_data_ny.copy(), 'NYSE')
normalized_london = normalizer.normalize_to_utc(test_data_london.copy(), 'LSE')
print("Примеры временных меток после нормализации:")
print(f"NYSE (UTC): {normalized_ny.index[:3]}")
print(f"LSE  (UTC): {normalized_london.index[:3]}")
# Выравниваем по общим торговым дням
aligned_data = normalizer.align_trading_days({
'NYSE': test_data_ny,
'LSE': test_data_london
}, method='intersection', agg='mean')   # берем средние цены за день
print("\nСовпадающие торговые дни:")
print(aligned_data['NYSE'].index[:5])
print(f"Количество совпадающих дней: {len(aligned_data['NYSE'])}")

Примеры временных меток после нормализации:
NYSE (UTC): DatetimeIndex(['2025-01-01 05:00:00+00:00', '2025-01-02 05:00:00+00:00',
'2025-01-03 05:00:00+00:00'],
dtype='datetime64[ns, UTC]', freq='D')
LSE  (UTC): DatetimeIndex(['2025-01-01 00:00:00+00:00', '2025-01-02 00:00:00+00:00',
'2025-01-03 00:00:00+00:00'],
dtype='datetime64[ns, UTC]', freq='D')
Совпадающие торговые дни:
DatetimeIndex(['2025-01-01 00:00:00+00:00', '2025-01-02 00:00:00+00:00',
'2025-01-03 00:00:00+00:00', '2025-01-04 00:00:00+00:00',
'2025-01-05 00:00:00+00:00'],
dtype='datetime64[ns, UTC]', freq='D')
Количество совпадающих дней: 99

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
class CorporateActionsAdjuster:
def __init__(self):
self.action_history = {}
def register_split(self, ticker, date, ratio):
"""Регистрация сплита акций"""
if ticker not in self.action_history:
self.action_history[ticker] = []
self.action_history[ticker].append({
'type': 'split',
'date': pd.to_datetime(date),
'ratio': ratio  # Например, 2.0 для сплита 2:1
})
def register_dividend(self, ticker, ex_date, amount):
"""Регистрация дивиденда"""
if ticker not in self.action_history:
self.action_history[ticker] = []
self.action_history[ticker].append({
'type': 'dividend',
'date': pd.to_datetime(ex_date),
'amount': amount
})
def get_nearest_trading_date(self, data, date):
"""Найти ближайший торговый день >= указанной даты"""
if date in data.index:
return date
future_dates = data.index[data.index >= date]
if len(future_dates) > 0:
return future_dates[0]
return None
def backward_adjust_prices(self, data, ticker):
"""Обратная корректировка цен на корпоративные действия"""
if ticker not in self.action_history:
return data.copy()
adjusted = data.copy()
actions = sorted(self.action_history[ticker], 
key=lambda x: x['date'], reverse=True)
for action in actions:
action_date = self.get_nearest_trading_date(adjusted, action['date'])
if action_date is None:
continue  # дата вне диапазона данных
mask = adjusted.index < action_date
if action['type'] == 'split':
split_ratio = action['ratio']
adjusted.loc[mask, ['Open','High','Low','Close']] /= split_ratio
adjusted.loc[mask, 'Volume'] *= split_ratio
elif action['type'] == 'dividend':
dividend_amount = action['amount']
close_price = adjusted.loc[action_date, 'Close']
adjustment_factor = 1 - (dividend_amount / close_price)
adjusted.loc[mask, ['Open','High','Low','Close']] *= adjustment_factor
return adjusted
def calculate_total_return(self, data, ticker, initial_shares=100):
"""Расчет total return с учетом реинвестирования дивидендов"""
if ticker not in self.action_history:
price_return = ((data['Close'].iloc[-1] / data['Close'].iloc[0]) - 1) * 100
return price_return, price_return
shares = initial_shares
cash = 0.0
actions = sorted(self.action_history[ticker], key=lambda x: x['date'])
for action in actions:
action_date = self.get_nearest_trading_date(data, action['date'])
if action_date is None:
continue
if action['type'] == 'split':
shares *= action['ratio']
elif action['type'] == 'dividend':
dividend_payment = shares * action['amount']
# Реинвестирование дивидендов
price_at_ex_date = data.loc[action_date, 'Close']
additional_shares = dividend_payment / price_at_ex_date
shares += additional_shares
initial_value = initial_shares * data['Close'].iloc[0]
final_value = shares * data['Close'].iloc[-1] + cash
total_return = ((final_value / initial_value) - 1) * 100
price_return = ((data['Close'].iloc[-1] / data['Close'].iloc[0]) - 1) * 100
return total_return, price_return
def detect_potential_splits(self, data, threshold=0.4):
"""Автоматическая детекция потенциальных сплитов"""
returns = data['Close'].pct_change()
volume_changes = data['Volume'].pct_change()
# Ищем дни с большим падением цены и ростом объема
potential_splits = returns[(returns < -threshold) & (volume_changes > 0.5)]
detected = []
for date, ret in potential_splits.items():
if date not in data.index:
continue
if len(data.loc[:date]) < 2:
continue
price_before = data.loc[:date, 'Close'].iloc[-2]
price_after = data.loc[date, 'Close']
# Проверяем распространенные коэффициенты сплита
common_ratios = [2.0, 3.0, 4.0, 5.0, 10.0, 0.5, 0.333, 0.25, 0.2]
for ratio in common_ratios:
if abs(price_after / price_before - 1/ratio) < 0.05:
detected.append({
'date': date,
'suspected_ratio': ratio,
'price_change': ret,
'volume_change': volume_changes[date]
})
break
return pd.DataFrame(detected)
# ДЕМО
adjuster = CorporateActionsAdjuster()
# Создаем синтетические данные
synthetic_data = pd.DataFrame({
'Open': 100 + np.random.randn(len(dates)).cumsum() * 0.2,
'High': 102 + np.random.randn(len(dates)).cumsum() * 0.2,
'Low': 98 + np.random.randn(len(dates)).cumsum() * 0.2,
'Close': 100 + np.random.randn(len(dates)).cumsum() * 0.2,
'Volume': np.random.randint(1_000_000, 5_000_000, len(dates))
}, index=dates)
# Регистрируем корпоративные действия
adjuster.register_split('TEST', '2024-06-01', 2.0)     
adjuster.register_dividend('TEST', '2024-09-15', 0.1)  # Дивиденд (воскресенье → перенесётся на 2024-09-16)
# Применяем корректировку
adjusted_data = adjuster.backward_adjust_prices(synthetic_data, 'TEST')
# Сравнение до и после
comparison = pd.DataFrame({
'Original_Close': synthetic_data['Close'],
'Adjusted_Close': adjusted_data['Close'],
'Difference_%': ((adjusted_data['Close'] - synthetic_data['Close']) / 
synthetic_data['Close'] * 100)
})
print("Влияние корпоративных действий на исторические цены:")
print(comparison.iloc[[100, 150, 200, 250]].round(2))
# Расчет total return
total_ret, price_ret = adjuster.calculate_total_return(synthetic_data, 'TEST')
print(f"\nTotal Return (с реинвестированием): {total_ret:.2f}%")
print(f"Price Return (без дивидендов): {price_ret:.2f}%")
print(f"Дивидендная составляющая: {total_ret - price_ret:.2f}%")

Влияние корпоративных действий на исторические цены:
Original_Close  Adjusted_Close  Difference_%
2024-05-20          100.43           50.16        -50.05
2024-07-29          101.44          101.34         -0.10
2024-10-07          100.66          100.66          0.00
2024-12-16           98.65           98.65          0.00
Total Return (с реинвестированием): 98.61%
Price Return (без дивидендов): -0.79%
Дивидендная составляющая: 99.41%

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
pd.set_option('display.expand_frame_repr', False)
import warnings
warnings.filterwarnings('ignore')
class DataSynchronizer:
def __init__(self):
self.sync_methods = {
'forward_fill': self._forward_fill_sync,
'nearest': self._nearest_sync,
'interpolate': self._interpolate_sync
}
def _forward_fill_sync(self, high_freq_index, low_freq_data):
"""Forward fill - использует последнее доступное значение"""
# Reindex с forward fill
synced = low_freq_data.reindex(high_freq_index, method='ffill')
return synced
def _nearest_sync(self, high_freq_index, low_freq_data):
"""Nearest - использует ближайшее по времени значение"""
synced = low_freq_data.reindex(high_freq_index, method='nearest')
return synced
def _interpolate_sync(self, high_freq_index, low_freq_data):
"""Интерполяция для числовых данных"""
# Объединяем индексы
combined_index = high_freq_index.union(low_freq_data.index)
combined_index = combined_index.sort_values()
# Reindex и интерполяция
reindexed = low_freq_data.reindex(combined_index)
interpolated = reindexed.interpolate(method='time')
# Возвращаем только высокочастотные точки
synced = interpolated.reindex(high_freq_index)
return synced
def synchronize_multiple_frequencies(self, data_dict, target_freq='1D', method='forward_fill'):
"""
Синхронизация данных разных частот к целевой частоте
data_dict: словарь {name: DataFrame} с данными разных частот
target_freq: целевая частота ('1D', '1H', '5min' и т.д.)
method: метод синхронизации
"""
if not data_dict:
return pd.DataFrame()
# Определяем общий временной диапазон
min_date = min(df.index.min() for df in data_dict.values())
max_date = max(df.index.max() for df in data_dict.values())
# Создаем целевой индекс
target_index = pd.date_range(start=min_date, end=max_date, freq=target_freq)
# Синхронизируем каждый датафрейм
synced_data = {}
sync_func = self.sync_methods.get(method, self._forward_fill_sync)
for name, df in data_dict.items():
if isinstance(df, pd.Series):
synced_data[name] = sync_func(target_index, df)
elif isinstance(df, pd.DataFrame):
synced_df = pd.DataFrame(index=target_index)
for col in df.columns:
synced_df[f"{name}_{col}"] = sync_func(target_index, df[col])
synced_data[name] = synced_df
# Объединяем все в один DataFrame
result = pd.DataFrame(index=target_index)
for name, data in synced_data.items():
if isinstance(data, pd.Series):
result[name] = data
elif isinstance(data, pd.DataFrame):
result = pd.concat([result, data], axis=1)
return result
def apply_reporting_lag(self, fundamental_data, lag_days=45):
"""
Применение reporting lag к фундаментальным данным
Симулирует реальную задержку в доступности финансовых отчетов
"""
lagged = fundamental_data.copy()
lagged.index = lagged.index + pd.Timedelta(days=lag_days)
return lagged
def create_point_in_time_features(self, price_data, fundamental_data,
technical_indicators, reporting_lag=60):
"""
Создание point-in-time корректного набора признаков
"""
# Применяем reporting lag к фундаментальным данным
lagged_fundamental = self.apply_reporting_lag(fundamental_data, reporting_lag)
# Синхронизируем все к дневной частоте с forward fill
all_data = {
'price': price_data,
'fundamental': lagged_fundamental,
'technical': technical_indicators
}
synced = self.synchronize_multiple_frequencies(
all_data,
target_freq='1D',
method='forward_fill'
)
return synced
class CrossAssetSynchronizer:
"""Синхронизация данных между различными активами и рынками"""
def __init__(self):
self.market_hours = {
'US': {'open': '09:30', 'close': '16:00', 'tz': 'America/New_York'},
'EU': {'open': '08:00', 'close': '16:30', 'tz': 'Europe/London'},
'ASIA': {'open': '09:00', 'close': '15:00', 'tz': 'Asia/Tokyo'}
}
def align_trading_hours(self, data_dict, reference_market='US'):
"""
Выравнивание данных по торговым часам референсного рынка
"""
ref_tz = self.market_hours[reference_market]['tz']
aligned = {}
for asset, df in data_dict.items():
# Конвертируем в UTC если есть timezone
if df.index.tz is not None:
df_utc = df.copy()
df_utc.index = df_utc.index.tz_convert('UTC')
else:
df_utc = df.copy()
df_utc.index = df_utc.index.tz_localize('UTC')
# Затем конвертируем в референсный timezone
df_aligned = df_utc.copy()
df_aligned.index = df_aligned.index.tz_convert(ref_tz)
aligned[asset] = df_aligned
return aligned
def create_lagged_correlations(self, data_dict, max_lag=5):
"""
Расчет кросс-корреляций с учетом лагов между рынками
Полезно для стратегий, использующих lead-lag эффекты
"""
assets = list(data_dict.keys())
correlations = {}
for i, asset1 in enumerate(assets):
for asset2 in assets[i+1:]:
df1 = data_dict[asset1]
df2 = data_dict[asset2]
# Находим общие даты
common_dates = df1.index.intersection(df2.index)
if len(common_dates) < 30:
continue
s1 = df1.loc[common_dates, 'Close'].pct_change()
s2 = df2.loc[common_dates, 'Close'].pct_change()
# Рассчитываем корреляции для разных лагов
lag_corrs = {}
for lag in range(-max_lag, max_lag + 1):
if lag < 0: # asset1 лидирует corr = s1.iloc[:lag].corr(s2.iloc[-lag:]) elif lag > 0:
# asset2 лидирует
corr = s1.iloc[lag:].corr(s2.iloc[:-lag])
else:
# Синхронная корреляция
corr = s1.corr(s2)
lag_corrs[lag] = corr
correlations[f"{asset1}_{asset2}"] = lag_corrs
return pd.DataFrame(correlations).T
# Демонстрация синхронизации данных
def demo_data_synchronization():
np.random.seed(42)
# Создаем данные разных частот
# Дневные цены (высокая частота для примера)
daily_dates = pd.date_range('2024-01-01', '2024-12-31', freq='D')
daily_prices = pd.DataFrame({
'Close': 100 + np.random.randn(len(daily_dates)).cumsum() * 2,
'Volume': np.random.randint(1000000, 5000000, len(daily_dates))
}, index=daily_dates)
# Недельные технические индикаторы
weekly_dates = pd.date_range('2024-01-01', '2024-12-31', freq='W')
weekly_indicators = pd.DataFrame({
'Momentum': np.random.randn(len(weekly_dates)) * 5,
'Volatility': 10 + np.abs(np.random.randn(len(weekly_dates)) * 3)
}, index=weekly_dates)
# Квартальные фундаментальные данные
quarterly_dates = pd.date_range('2024-01-01', '2024-12-31', freq='Q')
quarterly_fundamental = pd.DataFrame({
'EPS': 2.5 + np.random.randn(len(quarterly_dates)) * 0.3,
'Revenue': 1000 + np.random.randn(len(quarterly_dates)) * 50
}, index=quarterly_dates)
print("=== Синхронизация данных разных частот ===\n")
# Инициализируем синхронайзер
synchronizer = DataSynchronizer()
# Синхронизируем к дневной частоте
data_dict = {
'weekly': weekly_indicators,
'quarterly': quarterly_fundamental
}
synced_data = synchronizer.synchronize_multiple_frequencies(
data_dict,
target_freq='1D',
method='forward_fill'
)
print(f"Исходные частоты данных:")
print(f"  Недельные индикаторы: {len(weekly_indicators)} строк")
print(f"  Квартальные фундаментальные: {len(quarterly_fundamental)} строк")
print(f"\nПосле синхронизации: {len(synced_data)} строк")
# Применяем reporting lag
lagged_fundamental = synchronizer.apply_reporting_lag(quarterly_fundamental, lag_days=60)
print(f"\n=== Reporting Lag ===")
print(f"Оригинальные даты фундаментальных данных:")
print(quarterly_fundamental.index.tolist())
print(f"\nС учетом 60-дневного лага:")
print(lagged_fundamental.index.tolist())
# Создаем point-in-time корректный датасет
pit_features = synchronizer.create_point_in_time_features(
daily_prices,
quarterly_fundamental,
weekly_indicators,
reporting_lag=60
)
print(f"\n=== Point-in-Time Features ===")
print(f"Всего признаков: {pit_features.shape[1]}")
print(f"Период: {pit_features.index.min()} - {pit_features.index.max()}")
print(pit_features.head())
# Демонстрация кросс-ассет синхронизации
print(f"\n=== Cross-Asset Synchronization ===")
# Создаем данные для разных рынков
us_data = daily_prices.copy()
us_data.index = us_data.index.tz_localize('America/New_York')
eu_data = daily_prices.copy() * 1.1  # Немного другие цены
eu_data.index = eu_data.index.tz_localize('Europe/London')
cross_sync = CrossAssetSynchronizer()
aligned = cross_sync.align_trading_hours({'US': us_data, 'EU': eu_data}, reference_market='US')
print(f"US данные timezone: {aligned['US'].index.tz}")
print(f"EU данные timezone: {aligned['EU'].index.tz}")
# Расчет lagged correlations
lagged_corrs = cross_sync.create_lagged_correlations({'US': us_data, 'EU': eu_data}, max_lag=3)
if not lagged_corrs.empty:
print(f"\n=== Lagged Correlations ===")
print(lagged_corrs)
demo_data_synchronization()

=== Синхронизация данных разных частот ===
Исходные частоты данных:
Недельные индикаторы: 52 строк
Квартальные фундаментальные: 4 строк
После синхронизации: 360 строк
=== Reporting Lag ===
Оригинальные даты фундаментальных данных:
[Timestamp('2024-03-31 00:00:00'), Timestamp('2024-06-30 00:00:00'), Timestamp('2024-09-30 00:00:00'), Timestamp('2024-12-31 00:00:00')]
С учетом 60-дневного лага:
[Timestamp('2024-05-30 00:00:00'), Timestamp('2024-08-29 00:00:00'), Timestamp('2024-11-29 00:00:00'), Timestamp('2025-03-01 00:00:00')]
=== Point-in-Time Features ===
Всего признаков: 6
Период: 2024-01-01 00:00:00 - 2025-03-01 00:00:00
=== Cross-Asset Synchronization ===
US данные timezone: America/New_York
EU данные timezone: America/New_York

import pandas as pd
import numpy as np
from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, Index
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime, timedelta
import pickle
import lzma
Base = declarative_base()
class OHLCVData(Base):
__tablename__ = 'ohlcv_data'
id = Column(Integer, primary_key=True)
ticker = Column(String(20), nullable=False)
timestamp = Column(DateTime, nullable=False)
open = Column(Float)
high = Column(Float)
low = Column(Float)
close = Column(Float)
volume = Column(Integer)
adjusted = Column(Float)  # Adjusted close
# Композитный индекс для быстрых запросов
__table_args__ = (
Index('idx_ticker_timestamp', 'ticker', 'timestamp'),
Index('idx_timestamp', 'timestamp'),
)
class TimeSeriesStorage:
def __init__(self, db_url='sqlite:///market_data.db', use_compression=True):
self.engine = create_engine(db_url, echo=False)
Base.metadata.create_all(self.engine)
Session = sessionmaker(bind=self.engine)
self.session = Session()
self.use_compression = use_compression
def store_dataframe(self, ticker, df, batch_size=1000):
"""Эффективная запись DataFrame в базу данных"""
records = []
for timestamp, row in df.iterrows():
record = OHLCVData(
ticker=ticker,
timestamp=timestamp.to_pydatetime() if isinstance(timestamp, pd.Timestamp) else timestamp,
open=float(row.get('Open', row.get('open', 0))),
high=float(row.get('High', row.get('high', 0))),
low=float(row.get('Low', row.get('low', 0))),
close=float(row.get('Close', row.get('close', 0))),
volume=int(row.get('Volume', row.get('volume', 0))),
adjusted=float(row.get('Adjusted', row.get('adjusted', row.get('Close', 0))))
)
records.append(record)
# Батчинг для эффективности
if len(records) >= batch_size:
self.session.bulk_save_objects(records)
self.session.commit()
records = []
# Сохраняем оставшиеся записи
if records:
self.session.bulk_save_objects(records)
self.session.commit()
def retrieve_dataframe(self, ticker, start_date=None, end_date=None):
"""Извлечение данных из базы в DataFrame"""
query = self.session.query(OHLCVData).filter(OHLCVData.ticker == ticker)
if start_date:
query = query.filter(OHLCVData.timestamp >= start_date)
if end_date:
query = query.filter(OHLCVData.timestamp <= end_date) query = query.order_by(OHLCVData.timestamp) results = query.all() if not results: return pd.DataFrame() data = { 'Open': [r.open for r in results], 'High': [r.high for r in results], 'Low': [r.low for r in results], 'Close': [r.close for r in results], 'Volume': [r.volume for r in results], 'Adjusted': [r.adjusted for r in results] } df = pd.DataFrame(data, index=[r.timestamp for r in results]) return df def get_available_tickers(self): """Получение списка всех доступных тикеров""" result = self.session.query(OHLCVData.ticker).distinct().all() return [r[0] for r in result] def get_date_range(self, ticker): """Получение диапазона дат для тикера""" result = self.session.query( OHLCVData.timestamp ).filter( OHLCVData.ticker == ticker ).order_by( OHLCVData.timestamp.asc() ).first() min_date = result[0] if result else None result = self.session.query( OHLCVData.timestamp ).filter( OHLCVData.ticker == ticker ).order_by( OHLCVData.timestamp.desc() ).first() max_date = result[0] if result else None return min_date, max_date class CompressedFileStorage: """Альтернативное хранилище с компрессией для больших датасетов""" def __init__(self, base_path='./data'): self.base_path = base_path import os os.makedirs(base_path, exist_ok=True) def store_dataframe(self, ticker, df): """Сохранение DataFrame с LZMA компрессией""" filepath = f"{self.base_path}/{ticker}.pkl.xz" # Pickle + LZMA дает компрессию ~10x для OHLCV данных with lzma.open(filepath, 'wb', preset=6) as f: pickle.dump(df, f) def retrieve_dataframe(self, ticker, start_date=None, end_date=None): """Загрузка DataFrame с фильтрацией по дате""" filepath = f"{self.base_path}/{ticker}.pkl.xz" try: with lzma.open(filepath, 'rb') as f: df = pickle.load(f) if start_date or end_date: mask = pd.Series(True, index=df.index) if start_date: mask &= df.index >= pd.to_datetime(start_date)
if end_date:
mask &= df.index <= pd.to_datetime(end_date) df = df[mask] return df except FileNotFoundError: return pd.DataFrame() def get_storage_stats(self, ticker): """Статистика хранения для тикера""" filepath = f"{self.base_path}/{ticker}.pkl.xz" try: import os compressed_size = os.path.getsize(filepath) # Загружаем для оценки uncompressed size df = self.retrieve_dataframe(ticker) uncompressed_size = df.memory_usage(deep=True).sum() compression_ratio = uncompressed_size / compressed_size if compressed_size > 0 else 0
return {
'compressed_bytes': compressed_size,
'uncompressed_bytes': uncompressed_size,
'compression_ratio': compression_ratio,
'rows': len(df)
}
except FileNotFoundError:
return None
# Демонстрация систем хранения
def demo_storage_systems():
# Создаем тестовые данные
dates = pd.date_range('2021-01-01', '2025-01-01', freq='D')
test_data = pd.DataFrame({
'Open': 100 + np.random.randn(len(dates)).cumsum(),
'High': 102 + np.random.randn(len(dates)).cumsum(),
'Low': 98 + np.random.randn(len(dates)).cumsum(),
'Close': 100 + np.random.randn(len(dates)).cumsum(),
'Volume': np.random.randint(1000000, 10000000, len(dates)),
'Adjusted': 100 + np.random.randn(len(dates)).cumsum()
}, index=dates)
# Тестируем SQL хранилище
print("=== SQL Storage Test ===")
sql_storage = TimeSeriesStorage(db_url='sqlite:///test_market_data.db')
import time
start = time.time()
sql_storage.store_dataframe('TEST', test_data)
write_time = time.time() - start
print(f"SQL Write time: {write_time:.3f}s")
start = time.time()
retrieved = sql_storage.retrieve_dataframe(
'TEST',
start_date=datetime(2024, 1, 1),
end_date=datetime(2024, 12, 31)
)
read_time = time.time() - start
print(f"SQL Read time (1 year): {read_time:.3f}s")
print(f"Retrieved {len(retrieved)} rows")
# Тестируем Compressed File Storage
print("\n=== Compressed File Storage Test ===")
file_storage = CompressedFileStorage()
start = time.time()
file_storage.store_dataframe('TEST', test_data)
write_time = time.time() - start
print(f"Compressed Write time: {write_time:.3f}s")
start = time.time()
retrieved = file_storage.retrieve_dataframe(
'TEST',
start_date=datetime(2024, 1, 1),
end_date=datetime(2024, 12, 31)
)
read_time = time.time() - start
print(f"Compressed Read time (1 year): {read_time:.3f}s")
stats = file_storage.get_storage_stats('TEST')
if stats:
print(f"\nStorage stats:")
print(f"Compressed size: {stats['compressed_bytes']/1024:.1f} KB")
print(f"Uncompressed size: {stats['uncompressed_bytes']/1024:.1f} KB")
print(f"Compression ratio: {stats['compression_ratio']:.1f}x")
demo_storage_systems()

=== SQL Storage Test ===
SQL Write time: 0.260s
SQL Read time (1 year): 0.008s
Retrieved 367 rows
=== Compressed File Storage Test ===
Compressed Write time: 0.028s
Compressed Read time (1 year): 0.007s
Storage stats:
Compressed size: 57.1 KB
Uncompressed size: 80.0 KB
Compression ratio: 1.4x

import pandas as pd
import numpy as np
from scipy import stats
from sklearn.ensemble import IsolationForest
import matplotlib.pyplot as plt
from typing import Dict, List, Tuple
class AnomalyDetector:
def __init__(self, z_threshold=3.5, contamination=0.01):
self.z_threshold = z_threshold
self.contamination = contamination
self.anomaly_log = []
def structural_validation(self, df):
"""Базовая структурная проверка OHLCV данных"""
issues = []
# Проверка положительности цен
for col in ['Open', 'High', 'Low', 'Close']:
if col in df.columns:
negative_mask = df[col] <= 0 if negative_mask.any(): issues.append({ 'type': 'negative_price', 'column': col, 'count': negative_mask.sum(), 'dates': df.index[negative_mask].tolist() }) # Проверка OHLC соотношений if all(col in df.columns for col in ['Open', 'High', 'Low', 'Close']): # High должен быть >= max(Open, Close)
high_invalid = df['High'] < df[['Open', 'Close']].max(axis=1)
if high_invalid.any():
issues.append({
'type': 'invalid_high',
'count': high_invalid.sum(),
'dates': df.index[high_invalid].tolist()
})
# Low должен быть <= min(Open, Close) low_invalid = df['Low'] > df[['Open', 'Close']].min(axis=1)
if low_invalid.any():
issues.append({
'type': 'invalid_low',
'count': low_invalid.sum(),
'dates': df.index[low_invalid].tolist()
})
# Проверка нулевых объемов
if 'Volume' in df.columns:
zero_volume = df['Volume'] == 0
if zero_volume.any():
issues.append({
'type': 'zero_volume',
'count': zero_volume.sum(),
'dates': df.index[zero_volume].tolist()
})
return issues
def detect_price_spikes(self, df, column='Close'):
"""Детектирование ценовых спайков через модифицированный Z-score"""
if column not in df.columns or len(df) < 10: return pd.Series(False, index=df.index) # Рассчитываем логарифмические доходности returns = np.log(df[column] / df[column].shift(1)) # Modified Z-score (более робастный к выбросам) median = returns.median() mad = np.abs(returns - median).median() modified_z_scores = 0.6745 * (returns - median) / mad # Детектируем аномалии is_anomaly = np.abs(modified_z_scores) > self.z_threshold
return is_anomaly
def detect_volume_anomalies(self, df):
"""Детектирование аномальных объемов торгов"""
if 'Volume' not in df.columns or len(df) < 20:
return pd.Series(False, index=df.index)
# Используем IQR метод для объемов
q1 = df['Volume'].quantile(0.25)
q3 = df['Volume'].quantile(0.75)
iqr = q3 - q1
lower_bound = q1 - 3 * iqr
upper_bound = q3 + 3 * iqr
is_anomaly = (df['Volume'] < lower_bound) | (df['Volume'] > upper_bound)
return is_anomaly
def detect_multivariate_anomalies(self, df):
"""Многомерное детектирование через Isolation Forest"""
features = []
feature_names = []
# Подготавливаем признаки
if 'Close' in df.columns:
returns = df['Close'].pct_change()
features.append(returns.fillna(0))
feature_names.append('returns')
# Волатильность (rolling std)
volatility = returns.rolling(window=20).std()
features.append(volatility.fillna(volatility.mean()))
feature_names.append('volatility')
if 'Volume' in df.columns:
# Нормализованный объем
volume_norm = (df['Volume'] - df['Volume'].mean()) / df['Volume'].std()
features.append(volume_norm.fillna(0))
feature_names.append('volume_norm')
if all(col in df.columns for col in ['High', 'Low', 'Close']):
# Relative range
rel_range = (df['High'] - df['Low']) / df['Close']
features.append(rel_range.fillna(0))
feature_names.append('relative_range')
if len(features) < 2: return pd.Series(False, index=df.index) # Создаем матрицу признаков X = np.column_stack(features) # Isolation Forest clf = IsolationForest( contamination=self.contamination, random_state=42, n_estimators=100 ) predictions = clf.fit_predict(X) is_anomaly = predictions == -1 return pd.Series(is_anomaly, index=df.index) def comprehensive_check(self, df, ticker='UNKNOWN'): """Комплексная проверка на все типы аномалий""" report = { 'ticker': ticker, 'total_rows': len(df), 'structural_issues': [], 'price_anomalies': 0, 'volume_anomalies': 0, 'multivariate_anomalies': 0, 'anomaly_dates': [] } # Структурная валидация structural = self.structural_validation(df) report['structural_issues'] = structural # Ценовые спайки price_anomalies = self.detect_price_spikes(df) report['price_anomalies'] = price_anomalies.sum() # Объемные аномалии volume_anomalies = self.detect_volume_anomalies(df) report['volume_anomalies'] = volume_anomalies.sum() # Многомерные аномалии multi_anomalies = self.detect_multivariate_anomalies(df) report['multivariate_anomalies'] = multi_anomalies.sum() # Объединяем все обнаруженные аномалии all_anomalies = price_anomalies | volume_anomalies | multi_anomalies report['anomaly_dates'] = df.index[all_anomalies].tolist() # Логируем для дальнейшего анализа self.anomaly_log.append(report) return report, all_anomalies def clean_data(self, df, anomaly_mask, method='interpolate'): """Очистка данных от обнаруженных аномалий""" cleaned = df.copy() if method == 'interpolate': # Интерполяция для аномальных значений for col in ['Open', 'High', 'Low', 'Close']: if col in cleaned.columns: cleaned.loc[anomaly_mask, col] = np.nan cleaned[col] = cleaned[col].interpolate(method='time') elif method == 'forward_fill': # Forward fill for col in ['Open', 'High', 'Low', 'Close']: if col in cleaned.columns: cleaned.loc[anomaly_mask, col] = np.nan cleaned[col] = cleaned[col].fillna(method='ffill') elif method == 'remove': # Полное удаление аномальных строк cleaned = cleaned[~anomaly_mask] # Для Volume используем forward fill if 'Volume' in cleaned.columns: cleaned.loc[anomaly_mask, 'Volume'] = np.nan cleaned['Volume'] = cleaned['Volume'].fillna(method='ffill') return cleaned def visualize_anomalies(self, df, anomaly_mask, ticker='UNKNOWN'): """Визуализация обнаруженных аномалий""" fig, axes = plt.subplots(3, 1, figsize=(14, 10)) # График цен с аномалиями axes[0].plot(df.index, df['Close'], color='#333333', linewidth=1, label='Close Price') axes[0].scatter(df.index[anomaly_mask], df['Close'][anomaly_mask], color='red', s=50, marker='x', label='Anomalies', zorder=5) axes[0].set_title(f'{ticker}: Price with Detected Anomalies', fontsize=12, pad=10) axes[0].set_ylabel('Price', fontsize=10) axes[0].legend(loc='best') axes[0].grid(True, alpha=0.3) # График доходностей returns = df['Close'].pct_change() axes[1].plot(df.index, returns, color='#666666', linewidth=0.8, alpha=0.7, label='Returns') axes[1].scatter(df.index[anomaly_mask], returns[anomaly_mask], color='red', s=50, marker='x', label='Anomalies', zorder=5) axes[1].axhline(y=0, color='black', linestyle='-', linewidth=0.5) axes[1].set_title('Returns with Anomalies', fontsize=12, pad=10) axes[1].set_ylabel('Return', fontsize=10) axes[1].legend(loc='best') axes[1].grid(True, alpha=0.3) # График объемов axes[2].bar(df.index, df['Volume'], color='#444444', width=1, alpha=0.6, label='Volume') axes[2].bar(df.index[anomaly_mask], df['Volume'][anomaly_mask], color='red', width=1, alpha=0.8, label='Anomalies', zorder=5) axes[2].set_title('Volume with Anomalies', fontsize=12, pad=10) axes[2].set_ylabel('Volume', fontsize=10) axes[2].legend(loc='best') axes[2].grid(True, alpha=0.3) plt.tight_layout() return fig # Демонстрация обнаружения аномалий def demo_anomaly_detection(): # Создаем синтетические данные с искусственными аномалиями np.random.seed(42) dates = pd.date_range('2024-01-01', '2025-01-01', freq='D') # Базовые данные close_prices = 100 + np.random.randn(len(dates)).cumsum() * 2 synthetic_data = pd.DataFrame({ 'Open': close_prices + np.random.randn(len(dates)) * 0.5, 'High': close_prices + np.abs(np.random.randn(len(dates))) * 1.5, 'Low': close_prices - np.abs(np.random.randn(len(dates))) * 1.5, 'Close': close_prices, 'Volume': np.random.randint(1000000, 5000000, len(dates)) }, index=dates) # Вводим аномалии # Flash crash synthetic_data.loc['2024-05-15', 'Close'] *= 0.7 synthetic_data.loc['2024-05-15', 'Low'] *= 0.65 # Ценовой спайк synthetic_data.loc['2024-08-20', 'Close'] *= 1.4 synthetic_data.loc['2024-08-20', 'High'] *= 1.45 # Аномальный объем synthetic_data.loc['2024-10-10', 'Volume'] *= 10 # Некорректные OHLC соотношения synthetic_data.loc['2024-11-05', 'High'] = synthetic_data.loc['2024-11-05', 'Low'] * 0.95 print("=== Анализ данных с аномалиями ===\n") # Инициализируем детектор detector = AnomalyDetector(z_threshold=3.0, contamination=0.02) # Комплексная проверка report, anomaly_mask = detector.comprehensive_check(synthetic_data, ticker='SYNTHETIC') print(f"Тикер: {report['ticker']}") print(f"Всего строк: {report['total_rows']}") print(f"\nСтруктурные проблемы: {len(report['structural_issues'])}") for issue in report['structural_issues']: print(f" - {issue['type']}: {issue['count']} случаев") print(f"\nСтатистические аномалии:") print(f" - Ценовые спайки: {report['price_anomalies']}") print(f" - Объемные аномалии: {report['volume_anomalies']}") print(f" - Многомерные аномалии: {report['multivariate_anomalies']}") print(f"\nВсего обнаружено уникальных дат с аномалиями: {len(report['anomaly_dates'])}") if report['anomaly_dates']: print("\nПримеры дат с аномалиями:") for date in report['anomaly_dates'][:5]: print(f" - {date}") # Очистка данных cleaned_data = detector.clean_data(synthetic_data, anomaly_mask, method='interpolate') print(f"\n=== Сравнение оригинальных и очищенных данных ===") comparison = pd.DataFrame({ 'Original_Close': synthetic_data['Close'], 'Cleaned_Close': cleaned_data['Close'], 'Difference': synthetic_data['Close'] - cleaned_data['Close'] }) # Показываем только строки с изменениями changed = comparison[comparison['Difference'].abs() > 0.01]
if not changed.empty:
print(f"\nИзмененные значения ({len(changed)} строк):")
print(changed.head(10))
# Визуализация
fig = detector.visualize_anomalies(synthetic_data, anomaly_mask, ticker='SYNTHETIC')
plt.savefig('anomaly_detection_demo.png', dpi=100, bbox_inches='tight')
print("\nГрафик сохранен как 'anomaly_detection_demo.png'")
demo_anomaly_detection()

=== Анализ данных с аномалиями ===
Тикер: SYNTHETIC
Всего строк: 367
Структурные проблемы: 2
- invalid_high: 42 случаев
- invalid_low: 42 случаев
Статистические аномалии:
- Ценовые спайки: 8
- Объемные аномалии: 1
- Многомерные аномалии: 8
Всего обнаружено уникальных дат с аномалиями: 12
Примеры дат с аномалиями:
- 2024-03-15 00:00:00
- 2024-04-23 00:00:00
- 2024-05-15 00:00:00
- 2024-05-16 00:00:00
- 2024-05-26 00:00:00
=== Сравнение оригинальных и очищенных данных ===
Измененные значения (12 строк):
Original_Close  Cleaned_Close  Difference
2024-03-15       82.757308      86.198955   -3.441648
2024-04-23       79.398322      76.742719    2.655603
2024-05-15       55.662116      76.713852  -21.051737
2024-05-16       77.950801      77.010265    0.940536
2024-05-26       76.305001      76.221495    0.083506
2024-05-31       76.495822      75.469349    1.026473
2024-07-28       95.964676      92.682835    3.281841
2024-08-20      137.046570      98.885181   38.161389
2024-08-21       96.587206      99.971098   -3.383892
2024-08-22      100.875095     101.057015   -0.181921
/tmp/ipython-input-150334835.py:205: FutureWarning: Series.fillna with 'method' is deprecated and will raise in a future version. Use obj.ffill() or obj.bfill() instead.
cleaned['Volume'] = cleaned['Volume'].fillna(method='ffill')
График сохранен как 'anomaly_detection_demo.png'

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from collections import deque
import smtplib
from email.mime.text import MIMEText
from typing import Dict, List, Optional
import json
import logging
class DataQualityMonitor:
def __init__(self, alert_thresholds=None):
self.alert_thresholds = alert_thresholds or {
'missing_data_pct': 5.0,       # % тикеров с пропусками
'staleness_hours': 24,         # Максимальная задержка данных
'anomaly_pct': 2.0,            # % аномальных записей
'api_error_rate': 10.0,        # % неуспешных запросов
'correlation_deviation': 0.3   # Отклонение от исторической корреляции
}
self.metrics_history = deque(maxlen=1000)
self.alerts_log = []
logging.basicConfig(level=logging.INFO)
self.logger = logging.getLogger(__name__)
def check_data_completeness(self, expected_tickers, received_data):
"""Проверка полноты данных"""
expected_set = set(expected_tickers)
received_set = set(received_data.keys())
missing = expected_set - received_set
unexpected = received_set - expected_set
missing_pct = (len(missing) / len(expected_set)) * 100 if expected_set else 0
metric = {
'timestamp': datetime.now(),
'type': 'completeness',
'total_expected': len(expected_set),
'total_received': len(received_set),
'missing_count': len(missing),
'missing_pct': missing_pct,
'missing_tickers': list(missing)[:10],  # Первые 10 для лога
'unexpected_tickers': list(unexpected)[:10]
}
self.metrics_history.append(metric)
if missing_pct > self.alert_thresholds['missing_data_pct']:
self._trigger_alert('completeness', metric, 
f"Высокий процент пропущенных тикеров: {missing_pct:.1f}%")
return metric
def check_data_staleness(self, data_dict, current_time=None):
"""Проверка свежести данных"""
current_time = current_time or datetime.now()
staleness_stats = {}
for ticker, df in data_dict.items():
if df.empty:
staleness_stats[ticker] = {'hours': float('inf'), 'status': 'empty'}
continue
latest_timestamp = df.index.max()
# Убираем timezone для расчета разницы
if latest_timestamp.tz is not None:
latest_timestamp = latest_timestamp.tz_localize(None)
if isinstance(current_time, pd.Timestamp):
current_time_naive = current_time.tz_localize(None) if current_time.tz else current_time
else:
current_time_naive = current_time
staleness = current_time_naive - latest_timestamp
staleness_hours = staleness.total_seconds() / 3600
status = 'fresh' if staleness_hours < self.alert_thresholds['staleness_hours'] else 'stale' staleness_stats[ticker] = {'hours': staleness_hours, 'status': status} stale_count = sum(1 for s in staleness_stats.values() if s['status'] == 'stale') stale_pct = (stale_count / len(staleness_stats)) * 100 if staleness_stats else 0 metric = { 'timestamp': datetime.now(), 'type': 'staleness', 'total_tickers': len(staleness_stats), 'stale_count': stale_count, 'stale_pct': stale_pct, 'avg_staleness_hours': np.mean([s['hours'] for s in staleness_stats.values() if s['hours'] != float('inf')]), 'max_staleness_hours': max([s['hours'] for s in staleness_stats.values()]) } self.metrics_history.append(metric) if stale_pct > self.alert_thresholds['missing_data_pct']:
self._trigger_alert('staleness', metric,
f"Высокий процент устаревших данных: {stale_pct:.1f}%")
return metric, staleness_stats
def check_correlation_consistency(self, current_data, historical_correlations, 
ticker_pairs=None):
"""Проверка консистентности корреляций между инструментами"""
if ticker_pairs is None:
# Выбираем несколько пар для мониторинга
tickers = list(current_data.keys())
ticker_pairs = [(tickers[i], tickers[i+1]) 
for i in range(min(3, len(tickers)-1))]
deviations = {}
for ticker1, ticker2 in ticker_pairs:
if ticker1 not in current_data or ticker2 not in current_data:
continue
df1 = current_data[ticker1]
df2 = current_data[ticker2]
if len(df1) < 30 or len(df2) < 30:
continue
# Находим общие даты
common_dates = df1.index.intersection(df2.index)
if len(common_dates) < 30: continue # Текущая корреляция (последние 30 дней) returns1 = df1.loc[common_dates, 'Close'].pct_change().tail(30) returns2 = df2.loc[common_dates, 'Close'].pct_change().tail(30) current_corr = returns1.corr(returns2) # Историческая корреляция pair_key = f"{ticker1}_{ticker2}" historical_corr = historical_correlations.get(pair_key, current_corr) deviation = abs(current_corr - historical_corr) deviations[pair_key] = { 'current': current_corr, 'historical': historical_corr, 'deviation': deviation } max_deviation = max([d['deviation'] for d in deviations.values()]) if deviations else 0 metric = { 'timestamp': datetime.now(), 'type': 'correlation', 'pairs_checked': len(deviations), 'max_deviation': max_deviation, 'deviations': deviations } self.metrics_history.append(metric) if max_deviation > self.alert_thresholds['correlation_deviation']:
self._trigger_alert('correlation', metric,
f"Значительное отклонение корреляции: {max_deviation:.3f}")
return metric
def check_api_health(self, api_stats):
"""Мониторинг здоровья API"""
total_requests = api_stats.get('total_requests', 0)
errors = api_stats.get('errors', 0)
error_rate = (errors / total_requests * 100) if total_requests > 0 else 0
avg_response_time = api_stats.get('avg_response_time', 0)
metric = {
'timestamp': datetime.now(),
'type': 'api_health',
'total_requests': total_requests,
'errors': errors,
'error_rate': error_rate,
'avg_response_time': avg_response_time,
'current_backoff': api_stats.get('current_backoff', 1.0)
}
self.metrics_history.append(metric)
if error_rate > self.alert_thresholds['api_error_rate']:
self._trigger_alert('api_health', metric,
f"Высокий уровень ошибок API: {error_rate:.1f}%")
return metric
def _trigger_alert(self, alert_type, metric, message):
"""Генерация алерта"""
alert = {
'timestamp': datetime.now(),
'type': alert_type,
'severity': self._determine_severity(alert_type, metric),
'message': message,
'metric': metric
}
self.alerts_log.append(alert)
self.logger.warning(f"ALERT [{alert_type}]: {message}")
# Здесь можно добавить отправку email, Slack, PagerDuty и т.д.
return alert
def _determine_severity(self, alert_type, metric):
"""Определение серьезности алерта"""
if alert_type == 'completeness':
missing_pct = metric['missing_pct']
if missing_pct > 20:
return 'critical'
elif missing_pct > 10:
return 'high'
else:
return 'medium'
elif alert_type == 'staleness':
max_hours = metric['max_staleness_hours']
if max_hours > 72:
return 'critical'
elif max_hours > 48:
return 'high'
else:
return 'medium'
elif alert_type == 'api_health':
error_rate = metric['error_rate']
if error_rate > 25:
return 'critical'
elif error_rate > 15:
return 'high'
else:
return 'medium'
return 'low'
def generate_health_report(self, lookback_hours=24):
"""Генерация отчета о состоянии системы"""
cutoff_time = datetime.now() - timedelta(hours=lookback_hours)
recent_metrics = [m for m in self.metrics_history 
if m['timestamp'] > cutoff_time]
report = {
'period': f'Last {lookback_hours} hours',
'generated_at': datetime.now(),
'total_checks': len(recent_metrics),
'alerts': len([a for a in self.alerts_log if a['timestamp'] > cutoff_time]),
'by_type': {}
}
# Агрегируем по типам проверок
for metric_type in ['completeness', 'staleness', 'correlation', 'api_health']:
type_metrics = [m for m in recent_metrics if m['type'] == metric_type]
if not type_metrics:
continue
if metric_type == 'completeness':
report['by_type']['completeness'] = {
'checks': len(type_metrics),
'avg_missing_pct': np.mean([m['missing_pct'] for m in type_metrics]),
'max_missing_pct': max([m['missing_pct'] for m in type_metrics])
}
elif metric_type == 'staleness':
report['by_type']['staleness'] = {
'checks': len(type_metrics),
'avg_stale_pct': np.mean([m['stale_pct'] for m in type_metrics]),
'max_staleness_hours': max([m['max_staleness_hours'] for m in type_metrics])
}
elif metric_type == 'api_health':
report['by_type']['api_health'] = {
'checks': len(type_metrics),
'avg_error_rate': np.mean([m['error_rate'] for m in type_metrics]),
'total_requests': sum([m['total_requests'] for m in type_metrics])
}
return report
def export_metrics(self, filepath='metrics_export.json'):
"""Экспорт метрик для анализа"""
export_data = {
'thresholds': self.alert_thresholds,
'metrics': [
{k: (v.isoformat() if isinstance(v, datetime) else v) 
for k, v in m.items()}
for m in list(self.metrics_history)
],
'alerts': [
{k: (v.isoformat() if isinstance(v, datetime) else v) 
for k, v in a.items() if k != 'metric'}
for a in self.alerts_log
]
}
with open(filepath, 'w') as f:
json.dump(export_data, f, indent=2)
return filepath
# Демонстрация системы мониторинга
def demo_monitoring_system():
np.random.seed(42)
# Создаем тестовые данные
expected_tickers = ['BABA', 'TSM', 'SHOP', 'SQ', 'AMD', 'MU', 'INTC', 'QCOM']
# Симулируем получение данных (с некоторыми пропусками)
received_data = {}
for ticker in expected_tickers[:-2]:  # Пропускаем последние 2
dates = pd.date_range('2024-01-01', '2024-10-01', freq='D')
received_data[ticker] = pd.DataFrame({
'Close': 100 + np.random.randn(len(dates)).cumsum() * 2,
'Volume': np.random.randint(1000000, 5000000, len(dates))
}, index=dates)
# Добавляем один устаревший тикер
old_dates = pd.date_range('2024-01-01', '2024-08-01', freq='D')
received_data['STALE'] = pd.DataFrame({
'Close': 100 + np.random.randn(len(old_dates)).cumsum(),
'Volume': np.random.randint(1000000, 5000000, len(old_dates))
}, index=old_dates)
print("=== Мониторинг качества данных ===\n")
# Инициализируем монитор
monitor = DataQualityMonitor()
# Проверка полноты
completeness = monitor.check_data_completeness(expected_tickers, received_data)
print(f"Проверка полноты данных:")
print(f"  Ожидалось: {completeness['total_expected']} тикеров")
print(f"  Получено: {completeness['total_received']} тикеров")
print(f"  Пропущено: {completeness['missing_count']} ({completeness['missing_pct']:.1f}%)")
if completeness['missing_tickers']:
print(f"  Пропущенные тикеры: {completeness['missing_tickers']}")
# Проверка свежести
staleness, stale_stats = monitor.check_data_staleness(received_data)
print(f"\nПроверка свежести данных:")
print(f"  Устаревших: {staleness['stale_count']} ({staleness['stale_pct']:.1f}%)")
print(f"  Средняя задержка: {staleness['avg_staleness_hours']:.1f} часов")
print(f"  Максимальная задержка: {staleness['max_staleness_hours']:.1f} часов")
# Проверка корреляций
historical_corrs = {
'BABA_TSM': 0.65,
'SHOP_SQ': 0.72,
'AMD_MU': 0.81
}
correlation = monitor.check_correlation_consistency(
received_data, 
historical_corrs,
ticker_pairs=[('BABA', 'TSM'), ('SHOP', 'SQ'), ('AMD', 'MU')]
)
print(f"\nПроверка корреляций:")
print(f"  Пар проверено: {correlation['pairs_checked']}")
print(f"  Макс. отклонение: {correlation['max_deviation']:.3f}")
# Симулируем API статистику
api_stats = {
'total_requests': 150,
'errors': 12,
'avg_response_time': 0.85,
'current_backoff': 1.2
}
api_health = monitor.check_api_health(api_stats)
print(f"\nСостояние API:")
print(f"  Всего запросов: {api_health['total_requests']}")
print(f"  Ошибок: {api_health['errors']} ({api_health['error_rate']:.1f}%)")
print(f"  Среднее время отклика: {api_health['avg_response_time']:.2f}s")
# Генерируем отчет
report = monitor.generate_health_report(lookback_hours=24)
print(f"\n=== Сводный отчет ===")
print(f"Период: {report['period']}")
print(f"Всего проверок: {report['total_checks']}")
print(f"Алертов: {report['alerts']}")
if monitor.alerts_log:
print(f"\n=== Активные алерты ===")
for alert in monitor.alerts_log[-5:]:
print(f"  [{alert['severity'].upper()}] {alert['type']}: {alert['message']}")
# Экспортируем метрики
export_path = monitor.export_metrics('data_quality_metrics.json')
print(f"\nМетрики экспортированы в: {export_path}")
demo_monitoring_system()

=== Мониторинг качества данных ===
Проверка полноты данных:
Ожидалось: 8 тикеров
Получено: 7 тикеров
Пропущено: 2 (25.0%)
Пропущенные тикеры: ['INTC', 'QCOM']
Проверка свежести данных:
Устаревших: 7 (100.0%)
Средняя задержка: 8966.9 часов
Максимальная задержка: 10221.7 часов
Проверка корреляций:
Пар проверено: 3
Макс. отклонение: 0.671
Состояние API:
Всего запросов: 150
Ошибок: 12 (8.0%)
Среднее время отклика: 0.85s
=== Сводный отчет ===
Период: Last 24 hours
Всего проверок: 4
Алертов: 3
=== Активные алерты ===
[CRITICAL] completeness: Высокий процент пропущенных тикеров: 25.0%
[CRITICAL] staleness: Высокий процент устаревших данных: 100.0%
[LOW] correlation: Значительное отклонение корреляции: 0.671
Метрики экспортированы в: data_quality_metrics.json