export_indicators.py 文件源码

def __output(self, stock, indicators):
        """history????indicator???????????

        Args:
            stock: ?????????
            indicators: ????indicator????
        """
        filename = '-'.join([i.__class__.__name__ for i in indicators])
        output_path = os.path.join(const.OUTPUT_INDICATOR_DIR,
                                   '{}-{}.csv'.format(stock.symbol.lower(), filename))
        self._prepare_directory(os.path.dirname(output_path))

        headers = ['Date', 'Open', 'High', 'Low', 'Close', 'Volume']
        for i in indicators:
            headers.append(i.name_with_args)

        with open(output_path, 'w', encoding=self.encoding, newline='') as fp:
            w = csv.writer(fp)
            w.writerow(headers)
            for idx, hist in enumerate(stock.histories):
                row = [hist.date.strftime('%Y-%m-%d'),
                       hist.open_price,
                       hist.high_price,
                       hist.low_price,
                       hist.close_price,
                       hist.volume]
                for i in indicators:
                    try:
                        val = i.get(idx)
                        if isinstance(val, float):
                            # numpy.float64???float????????????
                            # ?CSV????????????????????????
                            val = float(val)
                    except NoDataError:
                        val = ''
                    row.append(val)
                w.writerow(row)

        plogger.info('[{}] ????????????'.format(output_path))