def build_feed(instruments, fromYear, toYear, storage, frequency=bar.Frequency.DAY, timezone=None, skipErrors=False):
    logger = pyalgotrade.logger.getLogger("yahoofinance")
    ret = yahoofeed.Feed(frequency, timezone)

    if not os.path.exists(storage):
        logger.info("Creating %s directory" % (storage))
        os.mkdir(storage)

    for year in range(fromYear, toYear+1):
        for instrument in instruments:
            fileName = os.path.join(storage, "%s-%d-yahoofinance.csv" % (instrument, year))
            if not os.path.exists(fileName):
                logger.info("Downloading %s %d to %s" % (instrument, year, fileName))
                try:
                    if frequency == bar.Frequency.DAY:
                        download_daily_bars(instrument, year, fileName)
                    elif frequency == bar.Frequency.WEEK:
                        download_weekly_bars(instrument, year, fileName)
                    else:
                        raise Exception("Invalid frequency")
                except Exception, e:
                    if skipErrors:
                        logger.error(str(e))
                        continue
                    else:
                        raise e
            ret.addBarsFromCSV(instrument, fileName)

    def dispatchImpl(self, eventFilter):
        ret = False
        try:
            eventType, eventData = self.__wsClient.getQueue().get(True, Client.QUEUE_TIMEOUT)
            if eventFilter is not None and eventType not in eventFilter:
                return False

            ret = True
            if eventType == WSClient.ON_TICKER:
                self.__tickerEvent.emit(eventData)
            elif eventType == WSClient.ON_WALLET:
                self.__walletEvent.emit(eventData)
            elif eventType == WSClient.ON_TRADE:
                self.__tradeEvent.emit(eventData)
            elif eventType == WSClient.ON_USER_ORDER:
                self.__userOrderEvent.emit(eventData)
            elif eventType == WSClient.ON_RESULT:
                ret = False
                requestId, data = eventData
                self.__onResult(requestId, data)
            elif eventType == WSClient.ON_REMARK:
                ret = False
                requestId, data = eventData
                self.__onRemark(requestId, data)
            elif eventType == WSClient.ON_CONNECTED:
                self.__onConnected()
            elif eventType == WSClient.ON_DISCONNECTED:
                self.__onDisconnected()
            else:
                ret = False
                logger.error("Invalid event received to dispatch: %s - %s" % (eventType, eventData))
        except Queue.Empty:
            pass
        return ret

 def stop(self):
     try:
         if self.__thread is not None and self.__thread.is_alive():
             logger.info("Shutting down client.")
             self.__stream.disconnect()
     except Exception, e:
         logger.error("Error disconnecting stream: %s." % (str(e)))

def serve(barFeed, strategyParameters, address, port, batchSize=200):
    """Executes a server that will provide bars and strategy parameters for workers to use.

    :param barFeed: The bar feed that each worker will use to backtest the strategy.
    :type barFeed: :class:`pyalgotrade.barfeed.BarFeed`.
    :param strategyParameters: The set of parameters to use for backtesting. An iterable object where **each element is a tuple that holds parameter values**.
    :param address: The address to listen for incoming worker connections.
    :type address: string.
    :param port: The port to listen for incoming worker connections.
    :type port: int.
    :param batchSize: The number of strategy executions that are delivered to each worker.
    :type batchSize: int.
    :rtype: A :class:`Results` instance with the best results found or None if no results were obtained.
    """

    paramSource = base.ParameterSource(strategyParameters)
    resultSinc = base.ResultSinc()
    s = xmlrpcserver.Server(paramSource, resultSinc, barFeed, address, port, batchSize=batchSize)
    logger.info("Starting server")
    s.serve()
    logger.info("Server finished")

    ret = None
    bestResult, bestParameters = resultSinc.getBest()
    if bestResult is not None:
        logger.info("Best final result %s with parameters %s" % (bestResult, bestParameters.args))
        ret = Results(bestParameters.args, bestResult)
    else:
        logger.error("No results. All jobs failed or no jobs were processed.")
    return ret

    def dispatchImpl(self, eventFilter):
        try:
            eventType, eventData = self.__queue.get(True, Client.QUEUE_TIMEOUT)
            if eventFilter is not None and eventType not in eventFilter:
                return

            if eventType == WSClient.ON_TICKER:
                self.__tickerEvent.emit(eventData)
            elif eventType == WSClient.ON_TRADE:
                self.__tradeEvent.emit(eventData)
            elif eventType == WSClient.ON_USER_ORDER:
                self.__userOrderEvent.emit(eventData)
            elif eventType == WSClient.ON_RESULT:
                requestId, result = eventData
                logger.info("Result: %s - %s" % (requestId, result))
            elif eventType == WSClient.ON_REMARK:
                requestId, data = eventData
                logger.info("Remark: %s - %s" % (requestId, data))
            elif eventType == WSClient.ON_CONNECTED:
                self.__onConnected()
            elif eventType == WSClient.ON_DISCONNECTED:
                self.__onDisconnected()
            else:
                logger.error("Invalid event received to dispatch: %s - %s" % (eventType, eventData))
        except Queue.Empty:
            pass

    def doCall(self):
        for identifier in self.__identifiers:
            try:
                trades = api.get_trades() #identifier
                trades.reverse()
                for barDict in trades:
                    bar = {}
                    trade = TradeBar(barDict)
                    bar[identifier] = trade
                    tid = trade.getTradeId()
                    if tid > self.last_tid:
                        self.last_tid = tid
                        self.__queue.put((
                            TradesAPIThread.ON_TRADE, bar
                        ))
                orders = api.get_orderbook(identifier)
                if len(orders['bids']) and len(orders['asks']):
                    best_ask = orders['asks'][0]
                    best_bid = orders['bids'][0]
                    #last_update = self.last_orderbook_ts + 1;

                       # max(
                       # best_ask['timestamp'], best_bid['timestamp']
                    #)
                    # if last_update > self.last_orderbook_ts:
                    #     self.last_orderbook_ts = last_update
                    self.__queue.put((
                        TradesAPIThread.ON_ORDER_BOOK_UPDATE,
                        {
                            'bid': float(best_bid[0]),
                            'ask': float(best_ask[0])
                        }
                    ))
            except api.MercadobitcoinError, e:
                logger.error(e)

 def addBarsFromCSV(self,instrument='', period='D', timeFrom = None, timeTo = None):
     '''mid
     添加一个symbol的历史数据到对象
     '''
     import pandas as pd
     import pyalgotrade.logger        
     dataCenter = self.dataCenter            
     logger = pyalgotrade.logger.getLogger("tushare")
 
     '''
     if not os.path.exists(storage):
         logger.info("Creating %s directory" % (storage))
         os.mkdir(storage) 
         
      
     '''
     if(not dataCenter.localStorage.exists(instrument,period)):
         logger.info("Downloading %s from %s to %s" % (instrument, str(timeFrom),str(timeTo)))
         try:
             if (period in self.dataCenter.localStorage.periods.keys()):
                 if dataCenter.downloadAndStoreKDataByCode(code = instrument,timeFrom = timeFrom,timeTo = timeTo,period=period):
                     logger.info("Downloading successed.")
                 else:
                     logger.info("Downloading failed.")
             else:
                 raise Exception("Invalid period")
         except Exception, e:
             logger.error(str(e))
             raise e

 def stop(self):
     try:
         self.__stopped = True
         if self.__thread is not None and self.__thread.is_alive():
             logger.info("Shutting down MtGox client.")
             self.__wsClient.stopClient()
     except Exception, e:
         logger.error("Error shutting down MtGox client: %s" % (str(e)))

def main():
    fromYear = 2000
    toYear = 2012

    try:
        symbolsFile = os.path.join("..", "symbols", "merval.xml")
        symbolsxml.parse(symbolsFile, lambda stock: download_files_for_symbol(stock.getTicker(), fromYear, toYear))
    except Exception, e:
        logger.error(str(e))

def build_feed(sourceCode, tableCodes, fromYear, toYear, storage, frequency=bar.Frequency.DAY, timezone=None, skipErrors=False, noAdjClose=False, authToken=None):
    """Build and load a :class:`pyalgotrade.barfeed.quandlfeed.Feed` using CSV files downloaded from Quandl.
    CSV files are downloaded if they haven't been downloaded before.

    :param sourceCode: The dataset source code.
    :type sourceCode: string.
    :param tableCodes: The dataset table codes.
    :type tableCodes: list.
    :param fromYear: The first year.
    :type fromYear: int.
    :param toYear: The last year.
    :type toYear: int.
    :param storage: The path were the files will be loaded from, or downloaded to.
    :type storage: string.
    :param frequency: The frequency of the bars. Only **pyalgotrade.bar.Frequency.DAY** or **pyalgotrade.bar.Frequency.WEEK**
        are supported.
    :param timezone: The default timezone to use to localize bars. Check :mod:`pyalgotrade.marketsession`.
    :type timezone: A pytz timezone.
    :param skipErrors: True to keep on loading/downloading files in case of errors.
    :type skipErrors: boolean.
    :param noAdjClose: True if the instruments don't have adjusted close values.
    :type noAdjClose: boolean.
    :param authToken: Optional. An authentication token needed if you're doing more than 50 calls per day.
    :type authToken: string.
    :rtype: :class:`pyalgotrade.barfeed.quandlfeed.Feed`.
    """

    logger = pyalgotrade.logger.getLogger("quandl")
    ret = quandlfeed.Feed(frequency, timezone)
    if noAdjClose:
        ret.setNoAdjClose()

    if not os.path.exists(storage):
        logger.info("Creating %s directory" % (storage))
        os.mkdir(storage)

    for year in range(fromYear, toYear+1):
        for tableCode in tableCodes:
            fileName = os.path.join(storage, "%s-%s-%d-quandl.csv" % (sourceCode, tableCode, year))
            if not os.path.exists(fileName):
                logger.info("Downloading %s %d to %s" % (tableCode, year, fileName))
                try:
                    if frequency == bar.Frequency.DAY:
                        download_daily_bars(sourceCode, tableCode, year, fileName, authToken)
                    elif frequency == bar.Frequency.WEEK:
                        download_weekly_bars(sourceCode, tableCode, year, fileName, authToken)
                    else:
                        raise Exception("Invalid frequency")
                except Exception as e:
                    if skipErrors:
                        logger.error(str(e))
                        continue
                    else:
                        raise e
            ret.addBarsFromCSV(tableCode, fileName)
    return ret

def main():
    try:
        writer = symbolsxml.Writer()
        for symbol in open("merval-symbols.txt", "r"):
            symbol = symbol.strip()
            process_symbol(writer, symbol)
        logger.info("Writing merval.xml")
        writer.write("merval.xml")
    except Exception, e:
        logger.error(str(e))

 def getNextBars(self):
     ret = None
     try:
         eventType, eventData = self.__queue.get(True, LiveFeed.QUEUE_TIMEOUT)
         if eventType == GetBarThread.ON_BARS:
             ret = eventData
         else:
             logger.error("Invalid event received: %s - %s" % (eventType, eventData))
     except queue.Empty:
         pass
     return ret

    def doCall(self):
        endDateTime = self.__nextBarClose
        self.__updateNextBarClose()
        bar_dict = {}

        for identifier in self._identifiers:
            try:
                if not self._tickDSDict[identifier].empty():
                    bar_dict[identifier] = build_bar(to_market_datetime(endDateTime), self._tickDSDict[identifier])
            except Exception, e:
                logger.error(e)

 def __onDisconnected(self):
     logger.error("Disconnection detected")
     if self.__enableReconnection:
         initialized = False
         while not self.__stopped and not initialized:
             logger.info("Reconnecting")
             initialized = self.__initializeClient()
             if not initialized:
                 time.sleep(5)
     else:
         self.__stopped = True