# 基础功能安装
pip install mootdx

# 完整功能安装（含所有扩展模块）
pip install 'mootdx[all]'

# 源码安装（获取最新开发版本）
git clone https://gitcode.com/GitHub_Trending/mo/mootdx
cd mootdx
pip install -e .

from mootdx.reader import Reader
import pandas as pd
import talib as ta

# 初始化本地数据读取器
reader = Reader.factory(market='std', tdxdir='/path/to/通达信数据目录')

# 定义股票列表和时间范围
stock_codes = ['600036', '601318', '000858']
start_date = '20200101'
end_date = '20231231'

# 批量获取数据并计算指标
all_data = {}
for code in stock_codes:
    # 获取日线数据
    df = reader.daily(symbol=code, start=start_date, end=end_date)
    
    # 计算MACD指标
    df['MACD'], df['MACDsignal'], df['MACDhist'] = ta.MACD(
        df['close'], fastperiod=12, slowperiod=26, signalperiod=9
    )
    
    # 计算RSI指标
    df['RSI'] = ta.RSI(df['close'], timeperiod=14)
    
    all_data[code] = df

# 合并数据并保存
combined_data = pd.concat(all_data, keys=stock_codes)
combined_data.to_pickle('historical_data_with_indicators.pkl')

from mootdx.quotes import Quotes
import time
from datetime import datetime
import pandas as pd

class MarketMonitor:
    def __init__(self):
        self.client = Quotes.factory(market='std')
        self.watch_list = ['600036', '000001', '399001']
        self.price_history = {code: [] for code in self.watch_list}
        
    def get_realtime_data(self):
        """获取实时行情数据"""
        data = {}
        for code in self.watch_list:
            quote = self.client.quote(symbol=code)
            if quote is not None and not quote.empty:
                data[code] = {
                    'price': quote['price'].values[0],
                    'change': quote['change'].values[0],
                    'volume': quote['volume'].values[0],
                    'time': datetime.now()
                }
                self.price_history[code].append(data[code])
        return data
        
    def check_alert_conditions(self, data):
        """检查预警条件"""
        alerts = []
        for code, info in data.items():
            # 价格波动超过3%触发预警
            if abs(info['change']) > 3:
                alerts.append(f"股票 {code} 价格波动超过3%: {info['change']}%")
        return alerts
        
    def run_monitor(self, interval=60):
        """运行监控系统"""
        print(f"开始市场监控，刷新间隔: {interval}秒")
        try:
            while True:
                current_data = self.get_realtime_data()
                alerts = self.check_alert_conditions(current_data)
                
                if alerts:
                    print(f"\n【{datetime.now()}】预警信息:")
                    for alert in alerts:
                        print(f"- {alert}")
                
                # 每小时保存一次历史数据
                if datetime.now().minute == 0 and datetime.now().second < 10:
                    for code in self.watch_list:
                        df = pd.DataFrame(self.price_history[code])
                        df.to_csv(f"{code}_price_history.csv", index=False)
                    print(f"\n【{datetime.now()}】历史数据已保存")
                    
                time.sleep(interval)
                
        except KeyboardInterrupt:
            print("\n监控系统已停止")

# 运行监控
monitor = MarketMonitor()
monitor.run_monitor(interval=30)  # 每30秒刷新一次

from mootdx.financial import Financial
import pandas as pd
import numpy as np

class FundamentalStrategy:
    def __init__(self):
        self.financial = Financial()
        
    def get_financial_indicators(self, code, year, quarter):
        """获取财务指标数据"""
        # 获取资产负债表
        balance_sheet = self.financial.balance(symbol=code, year=year, quarter=quarter)
        
        # 获取利润表
        income_statement = self.financial.income(symbol=code, year=year, quarter=quarter)
        
        # 获取现金流量表
        cash_flow = self.financial.cashflow(symbol=code, year=year, quarter=quarter)
        
        return {
            'balance': balance_sheet,
            'income': income_statement,
            'cashflow': cash_flow
        }
        
    def calculate_financial_ratios(self, financial_data):
        """计算关键财务比率"""
        balance = financial_data['balance']
        income = financial_data['income']
        
        ratios = {}
        
        # 流动比率 = 流动资产 / 流动负债
        current_assets = balance[balance['code'] == '流动资产合计']['value'].values[0]
        current_liabilities = balance[balance['code'] == '流动负债合计']['value'].values[0]
        ratios['current_ratio'] = current_assets / current_liabilities
        
        # 资产负债率 = 总负债 / 总资产
        total_liabilities = balance[balance['code'] == '负债合计']['value'].values[0]
        total_assets = balance[balance['code'] == '资产总计']['value'].values[0]
        ratios['debt_ratio'] = total_liabilities / total_assets
        
        # 毛利率 = (营业收入 - 营业成本) / 营业收入
        revenue = income[income['code'] == '营业收入']['value'].values[0]
        cost = income[income['code'] == '营业成本']['value'].values[0]
        ratios['gross_margin'] = (revenue - cost) / revenue
        
        return ratios
        
    def screen_stocks(self, stock_pool, year, quarter, criteria):
        """基于财务指标筛选股票"""
        qualified_stocks = []
        
        for code in stock_pool:
            try:
                financial_data = self.get_financial_indicators(code, year, quarter)
                ratios = self.calculate_financial_ratios(financial_data)
                
                # 检查是否符合所有筛选条件
                meet_criteria = True
                for key, value in criteria.items():
                    if key not in ratios or not eval(f"{ratios[key]}{value}"):
                        meet_criteria = False
                        break
                        
                if meet_criteria:
                    qualified_stocks.append({
                        'code': code,
                        'ratios': ratios
                    })
                    
            except Exception as e:
                print(f"处理股票 {code} 时出错: {str(e)}")
                continue
                
        return qualified_stocks

# 使用示例
if __name__ == "__main__":
    strategy = FundamentalStrategy()
    
    # 定义筛选条件: 流动比率>1.5, 资产负债率<0.5, 毛利率>0.3
    criteria = {
        'current_ratio': '>1.5',
        'debt_ratio': '<0.5',
        'gross_margin': '>0.3'
    }
    
    # 股票池 (示例)
    stock_pool = ['600036', '601318', '000858', '000333', '600519']
    
    # 获取2022年第4季度财务数据并筛选
    results = strategy.screen_stocks(
        stock_pool=stock_pool,
        year=2022,
        quarter=4,
        criteria=criteria
    )
    
    print(f"符合条件的股票数量: {len(results)}")
    for result in results:
        print(f"股票代码: {result['code']}")
        print(f"财务比率: {result['ratios']}\n")

from mootdx.quotes import Quotes
from functools import lru_cache
import time
from datetime import timedelta

class CachedQuotes:
    def __init__(self, cache_timeout=300):
        """
        带缓存的行情数据获取器
        :param cache_timeout: 缓存超时时间(秒)，默认5分钟
        """
        self.client = Quotes.factory(market='std')
        self.cache_timeout = cache_timeout
        self.cache = {}
        
    def _is_cache_valid(self, key):
        """检查缓存是否有效"""
        if key not in self.cache:
            return False
        cache_time, _ = self.cache[key]
        return time.time() - cache_time < self.cache_timeout
        
    def get_cached_data(self, key):
        """获取缓存数据"""
        if self._is_cache_valid(key):
            return self.cache[key][1]
        return None
        
    def set_cache_data(self, key, data):
        """设置缓存数据"""
        self.cache[key] = (time.time(), data)
        
    def bars(self, symbol, frequency, start, count):
        """带缓存的K线数据获取"""
        key = f"bars:{symbol}:{frequency}:{start}:{count}"
        cached_data = self.get_cached_data(key)
        
        if cached_data is not None:
            return cached_data
            
        # 缓存未命中，从接口获取
        data = self.client.bars(symbol, frequency, start, count)
        
        # 设置缓存
        self.set_cache_data(key, data)
        return data
        
    @lru_cache(maxsize=128)
    def static_info(self, symbol):
        """带LRU缓存的静态信息获取"""
        # 静态信息变化较少，使用LRU缓存
        return self.client.stock_basic(symbol=symbol)

# 使用示例
if __name__ == "__main__":
    # 创建带缓存的行情客户端，缓存超时设为300秒
    cached_quotes = CachedQuotes(cache_timeout=300)
    
    # 第一次请求 - 无缓存
    start_time = time.time()
    data1 = cached_quotes.bars('600036', 9, 0, 100)
    print(f"首次请求耗时: {time.time() - start_time:.4f}秒")
    
    # 第二次请求 - 有缓存
    start_time = time.time()
    data2 = cached_quotes.bars('600036', 9, 0, 100)
    print(f"缓存请求耗时: {time.time() - start_time:.4f}秒")
    
    # 静态信息缓存
    print("\n静态信息缓存测试:")
    start_time = time.time()
    info1 = cached_quotes.static_info('600036')
    print(f"首次静态信息请求耗时: {time.time() - start_time:.4f}秒")
    
    start_time = time.time()
    info2 = cached_quotes.static_info('600036')
    print(f"缓存静态信息请求耗时: {time.time() - start_time:.4f}秒")

from mootdx.reader import Reader
import pandas as pd
from concurrent.futures import ThreadPoolExecutor, as_completed
import os
from queue import Queue
import threading

class DistributedDataCollector:
    def __init__(self, tdxdir, max_workers=4):
        self.tdxdir = tdxdir
        self.max_workers = max_workers
        self.result_queue = Queue()
        
    def worker(self, task_queue):
        """工作线程函数"""
        reader = Reader.factory(market='std', tdxdir=self.tdxdir)
        while not task_queue.empty():
            try:
                code, start_date, end_date = task_queue.get(block=False)
                data = reader.daily(symbol=code, start=start_date, end=end_date)
                self.result_queue.put((code, data))
            except Exception as e:
                print(f"处理股票 {code} 时出错: {str(e)}")
            finally:
                task_queue.task_done()
                
    def collect_data(self, stock_codes, start_date, end_date):
        """分布式数据采集主函数"""
        # 创建任务队列
        task_queue = Queue()
        for code in stock_codes:
            task_queue.put((code, start_date, end_date))
            
        # 创建工作线程
        threads = []
        for _ in range(min(self.max_workers, len(stock_codes))):
            thread = threading.Thread(target=self.worker, args=(task_queue,))
            thread.start()
            threads.append(thread)
            
        # 等待所有任务完成
        task_queue.join()
        
        # 收集结果
        results = {}
        while not self.result_queue.empty():
            code, data = self.result_queue.get()
            results[code] = data
            
        return results

# 使用示例
if __name__ == "__main__":
    # 配置参数
    tdx_data_dir = '/path/to/通达信数据目录'
    stock_list = [f"6000{i:02d}" for i in range(1, 50)]  # 生成股票代码列表
    start_date = '20200101'
    end_date = '20231231'
    
    # 创建分布式采集器，使用4个工作线程
    collector = DistributedDataCollector(tdxdir=tdx_data_dir, max_workers=4)
    
    # 开始采集
    print(f"开始采集 {len(stock_list)} 只股票数据...")
    start_time = time.time()
    all_data = collector.collect_data(stock_list, start_date, end_date)
    elapsed_time = time.time() - start_time
    
    print(f"采集完成，耗时: {elapsed_time:.2f}秒")
    print(f"成功获取 {len(all_data)} 只股票数据")
    
    # 保存结果
    output_dir = 'distributed_data'
    os.makedirs(output_dir, exist_ok=True)
    
    for code, data in all_data.items():
        data.to_csv(f"{output_dir}/{code}.csv", index=False)

from mootdx.quotes import Quotes
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class MootdxDataFeed:
    """mootdx数据适配器，用于连接量化交易平台"""
    
    def __init__(self):
        self.client = Quotes.factory(market='std')
        self.symbol_mapping = {
            # 交易平台代码: 通达信代码
            'SH600036': '600036',
            'SH601318': '601318',
            'SZ000858': '000858'
        }
        
    def get_bars(self, symbol, frequency, count):
        """
        获取K线数据，适配量化平台接口
        :param symbol: 交易平台代码
        :param frequency: 周期，1=1分钟, 5=5分钟, 60=1小时, 1440=日线
        :param count: 获取数量
        :return: 标准化的K线数据DataFrame
        """
        # 转换为通达信代码
        tdx_symbol = self.symbol_mapping.get(symbol, symbol.split('.')[-1])
        
        # 转换频率
        freq_map = {1: 8, 5: 9, 15: 10, 30: 11, 60: 12, 1440: 9}
        tdx_freq = freq_map.get(frequency, 9)  # 默认日线
        
        # 获取数据
        data = self.client.bars(symbol=tdx_symbol, frequency=tdx_freq, start=0, count=count)
        
        if data is None or data.empty:
            return pd.DataFrame()
            
        # 标准化列名
        data = data.rename(columns={
            'open': 'open',
            'close': 'close',
            'high': 'high',
            'low': 'low',
            'volume': 'volume',
            'amount': 'turnover',
            'datetime': 'date'
        })
        
        # 确保日期格式正确
        data['date'] = pd.to_datetime(data['date'])
        
        return data[['date', 'open', 'high', 'low', 'close', 'volume', 'turnover']]
        
    def get_tick(self, symbol):
        """获取实时 tick 数据"""
        tdx_symbol = self.symbol_mapping.get(symbol, symbol.split('.')[-1])
        tick_data = self.client.quote(symbol=tdx_symbol)
        
        if tick_data is None or tick_data.empty:
            return None
            
        # 标准化tick数据
        return {
            'symbol': symbol,
            'last_price': tick_data['price'].values[0],
            'open_price': tick_data['open'].values[0],
            'high_price': tick_data['high'].values[0],
            'low_price': tick_data['low'].values[0],
            'volume': tick_data['volume'].values[0],
            'turnover': tick_data['amount'].values[0],
            'ask_price': [tick_data[f'ask{i}'].values[0] for i in range(1, 6)],
            'ask_volume': [tick_data[f'askvol{i}'].values[0] for i in range(1, 6)],
            'bid_price': [tick_data[f'bid{i}'].values[0] for i in range(1, 6)],
            'bid_volume': [tick_data[f'bidvol{i}'].values[0] for i in range(1, 6)],
            'update_time': datetime.now()
        }

# 量化平台集成示例 (伪代码)
"""
# 在量化交易平台中使用
from trading_platform import Strategy, Order

class MyStrategy(Strategy):
    def __init__(self):
        super().__init__()
        self.data_feed = MootdxDataFeed()
        self.symbols = ['SH600036', 'SH601318']
        
    def on_init(self):
        # 初始化策略
        for symbol in self.symbols:
            self.subscribe(symbol)
            
    def on_bar(self, bar):
        # 收到K线数据时触发
        symbol = bar.symbol
        # 获取历史数据
        hist_data = self.data_feed.get_bars(symbol, frequency=1440, count=60)
        
        if len(hist_data) < 60:
            return
            
        # 计算策略信号
        ma5 = hist_data['close'].rolling(5).mean().iloc[-1]
        ma20 = hist_data['close'].rolling(20).mean().iloc[-1]
        
        # 金叉买入信号
        if ma5 > ma20 and self.get_position(symbol).volume == 0:
            self.send_order(Order(symbol=symbol, volume=100, side='BUY'))
            
        # 死叉卖出信号
        elif ma5 < ma20 and self.get_position(symbol).volume > 0:
            self.send_order(Order(symbol=symbol, volume=100, side='SELL'))
"""