python 程序优化/并行?

[billgreen1]

我现在要将股票的 tick 数据转成频率为 1 分钟的数据。 数据是存在 amazon s3, 一个日期(文件夹)下面有 2400+只股票，以 zip 格式存储， zip 里面是 一个 csv 文件。 一个 csv 文件大约在 4000+行。

可以认为 csv 文件就是有一列是时间，大约间隔 3 秒，从 9:25:xx-15:0x:xx ，一列是价格。

现在采用的是笨办法，把文件下载到本地，已经下载了一个月的，大约 2G 左右。

主要写了 2 个函数，一个是读 zip 文件到 pandas 的 dataframe ，另一个是转换 dataframe 至 1min 的频率（这里是使用 dataframe 的 resample 函数)。

为了使用 resample ， dataframe 的 index 需要是 DatetimeIndex 格式。 我尝试了几种方法。

1. 使用 Pandas 的 DatetimeIndex ，很耗时。
2. strptime 也是很耗时。
3. 使用 datetime.datetime ， 类似如下调用：

def parse(time_string):
    year = int(time_string[0:4])
    month = ...
    ...
    dt = datetime.datetime(year,month,...)
    return dt
 

方法 3 比方法 1 、 2 快 3-30 倍左右。但是即使这样，还是很耗时。

def zip_to_dataframe(zipped_file):
    ...
    # index 之前是字符串格式
    df.index = df.index.map(parse) # 本行耗时占整个函数的 60%
    ...

第二个函数

def transform_dataframe(df):
    ...
    col_loc = transformed_df.columns.get_loc('open') #取得开盘价所在的列位置
    transformed_df.iloc[0,col_loc] = open_price # 本行耗时占整个函数的 45%
   ...

对于处理一个 zip 文件，两个函数各自耗时 100+毫秒左右。如果按照两个函数耗时 0.25 秒算， 0.25(秒) * 2400(一天有 2400 左右只股票) * 20(一个月大概 20 个左右的交易日) * 12(个月) / 3600(秒) = 40 小时！

即使我用并行，公司电脑只有双核.也要 20 小时。这只是 1 年的，如果要转换多年的。。。

我现在想到的:

1. 优化程序，减少耗时。对我来说很难想的出来了， datetime.datetime 后面应该是用的 C 函数吧？感觉很难提升了。 我想不明白为什么对一个单元格赋值会占用那么大的时间。

2. 把程序拷到其他电脑，一台电脑跑一年的数据，感觉这办法太土。。。。

3. 用 spark 分布式？没处理过，公司没用 spark ，我都是单机版玩玩，如果这是正途，我愿意试试。

4. 请多多赐教

[casparchen]

去找找有没有一分钟的数据呢

[loggerhead]

没仔细看，想到两个办法：
1. 换种语言实现，或者用 pypy
2. 起多个进程分别对不同的 zip 操作

[clino]

现在笔记本都有八核了。。。

[elitezhe]

两核不代表只能开两个线程啊，网络通信更多时间应该是花在了等待上，所以还是可以多开几个线程的啊

[bigtan]

我天天做这个，颇有心得，速度有办法提高数倍，可以联系我。

[mhycy]

方便放个样本和示例么？

[clino]

transformed_df 是什么对象? 它是最终你要的结果吗?

[evilic]

mark

[lecher]

Parse 可以优化，因为是连续时间，没必要每个数据序列都从字符串切割年月日这类的，改成预生成，取起始和结束两个时间段，直接把时间段字典构造好。
或者改成 singleton 模式，年月日时这类的直接缓存在类变量里面，有缓存的话就不要切割这段字符串了，直接取分秒的切割。
我倾向于第一种。

读 zip 文件和解包是磁盘 IO 和 CPU 密集型的活问，用 Python 估计优化提升的空间不大，要加速可能把用 SSD 或者把所有文件都读到内存再处理比较好。

[likuku]

python 嘛，一个 python 进程也只能用到 1 个 CPU 核，线程也只是一个进程内的。

我之前这么玩：
python 使用任务队列 queue ，根据可用的 cpu 核数量，开对应数量的 线程，
线程去 任务队列领任务，开子进程 subprocess 来处理任务。

如此这般，可以同时利用多个核。

虽然丑了点。

这么作有条件的：
任务可以互不关联的独立处理，没有依赖性。

或者，自己在外部解决进程间通讯（简易方法，可以用本地文件 /db/memcache/nosql 提供多进程读写状态）

[gamexg]

没细看，记得以前看过一篇说 python date 慢的文章，应该有用

strptime 真慢……
https://onebitbug.me/2013/09/27/python-strptime-is-super-slow/

[clino]

@likuku 还可以用 uwsgi 来跑多进程,这样比调子进程更好的是已经加载好了更省时间,比如我前两天刚发现的方方式 /t/265065 是用 rpc 来调用,当然也可以变成 http 之类的接口

不过楼主最多能利用 2 核 所以可能在考虑分布式,不过分布式的问题是不同的机器如何获取到待加工的文件,以及反馈处理过的数据

用 celery 之类的不知道合不合适

[billgreen1]

@elitezhe 数据已经被下载到本地了，如果耗时，也应该主要在 IO 上，不在网络通信上

[UnisandK]

https://www.packet.net/bare-metal/


开台按量的跑完删机器

[pimin]

可以丢点数据和 demo 出来给大家练手嘛
楼主这个问题就实际使用来说，并不是大问题
一年的数据 20 小时， 10 年也就 200 小时。
而且是一次性的，处理一次之后以后就没有需求了
你先找台机器跑起来，过几天就会解决了。
当然你可能还没把数据都下载完
如果有好的优化方案你后面处理也不用那么辛苦

[Zzzzzzzzz]

@likuku pandas 和 Python 的 IO 操作都是释放 GIL, 多线程可以跑在多核上的.

[lebowsk1s]

楼主做量化的？ anaconda 有个 mkl 发行版，还有个用到 GPU 的发行版，可以一试，还有，你的数据格式不给出来凭空让人看真不好理解