Python列表在分组大小上的平均值

3
我有N个不同长度的大型列表,每个列表中的每个值表示长度为25的固定窗口内的信号。也就是说,我每隔25秒/碱基等取信号的平均值,并将该值存储在列表中。
我对不同实验/设备进行此操作,运行时间不同(都是25的倍数)。例如,列表1运行1000次,列表1中有1000/25=40个值,列表2运行1025分钟,列表2中有1025/25=41个值,列表3运行2525次,列表3中有2525/25=101个值,以此类推...
现在,为了比较,我想将每个列表重新调整到相同数量的箱中,假设为40个箱。
实际上,list1resized的长度将为40,其值不会改变,因为1000/40=25。list2resized将从41个值的长度变为40个值的长度,list3将从101个值的长度变为40个值的长度(即所有列表现在都具有相同的大小)。
现在问题来了。如何通过对适当的箱子进行加权平均值来将每个列表调整为固定长度的40?
下面是一个示例,可以澄清这个问题。
list1 = [4.8, 6.9, ...]  #40 values for the 1000 run
list2 = [5.6, 7.8, 8.9, 13.4, ...] #41 values for the 1025 run
list3 = [4.1, 5.6, 10.3, 9.8, 40, 30, 21.4, 3, 2,...] #101 values for the 2525 run

现在,调整大小后的列表应该看起来像这样:
list1resized = [4.8*25/25, 6.9*25/25,...] #40 values for the 1000 run
list2resized = [(5.6*25+7.8*0.625)/25.625, (7.8*24.375+8.9*1.275)/25.625, (23.725*8.9+1.9*13.4)/25.625,...] # 40 values, averaged accordingly, for the 1025 run
list3resized = [(4.1*25+5.6*25+10.3*13.125)/(63.125), (10.3*11.875+9.8*25+40*25+30*1.25)/(63.125),...] # 40 values, averaged accordingly, for the 2525 run

为了获得每个重新调整大小的列表元素的平均值,我们对新的调整大小的容器进行了加权平均(即对于list1取1000/40=25个平均值,对于list2取1025/40=25.625个平均值,对于list3取2525/40=63.125个平均值等)。也就是说,使用我用于加权平均的公式。
list1resized = [4.8*25/25, 6.9*25/25,...] #40 values for the 1000 run
list2resized = [(5.6*25+7.8*0.625)/25.625, (7.8*24.375+8.9*(25.65-24.375))/(25.625), (23.725*8.9+(25.625-23.725)*13.4)/(25.625),...] # 40 values, averaged accordingly, for the 1025 run
list3resized = [(4.1*25+5.6*25+10.3*13.125)/(63.125), (10.3*(25-13.125)+9.8*25+40*25+30*(63.125-25*3+13.125)))/(63.125),...] # 40 values, averaged accordingly, for the 2525 run

您可以看到,这可能会变得混乱且难以处理,但我正在寻找一个Pythonic、优雅且快速的解决方案来解决这个问题。

我需要对许多列表执行此操作多次,因此考虑运行时间很重要。

不确定您是否有任何想法,但帮助将不胜感激。

谢谢。


我其实对如何在Python中编写这样的框架一无所知。我正在为每次需要取不同的子列表以及它们的长度会发生变化而苦恼...实际上不确定该怎么做。 - Dnaiel
如果你想让你的问题集中在最好的数学算法上,那么[scicomp.SE]可能是一个很好的问答平台。然后你可以在这里询问如何在Python中实现该算法。 - David Z
我很高兴其他人似乎理解了,并且你显然得到了所需的帮助。对我来说,这就像泥潭一样不清楚。我可以看出你是如何得出25、25.625和63.125的;但除此之外,我完全迷失了。 - John Y
3个回答

3
这个时髦(也许是)的解决方案怎么样?
首先是测量列表...
l = [5.6, 7.8, 8.9, 13.4]

将每个测量值复制25次(每秒一次...)

l1 = [item for sublist in [list(itertools.repeat(k,25)) for k in l] for item in sublist]

每秒钟进行归一化:

l2 = map(lambda x: x / 25., l1)

请参考以下这个SO帖子上的函数(已复制),该函数将列表切片为n个几乎相等的分区:

Python: Slicing a list into n nearly-equal-length partitions

def partition(lst, n):
    division = len(lst) / float(n)
    return [ lst[int(round(division * i)): int(round(division * (i + 1)))] for i in xrange(n) ]

定义您的列表的新长度

new_len = 2

将每秒列表分割成所需的段数:
l3 = partition(l2, new_len)

对于每个分区,对每秒的值进行求和

l4 = map(sum, l3)

针对列表大小的差异进行归一化处理

l5 = map(lambda x: x * new_len / float(len(l)), l4)

欣赏结果:

print l5

2
简而言之,[sum(t) * new_len / float(len(l)) for t in partition(itertools.chain(itertools.repeat(k/25., 25) for k in l), new_len)](大概就是这样) - David Z
我曾经考虑过将事物分割成秒,但老实说,在我看来,使用秒作为时间单位完全是不必要的。基于比率的解决方案完全忽略了“秒”参数(因为秒本身就是列表长度的函数),这更符合数学上的意义。 - kreativitea
@juniper 我认为找到一个好的解决方案的关键是找到一种生成子列表系数的方法。每个子列表系数的总和等于加权平均比率--[(4.1*25+5.6*25+10.3*13.125)/(63.125),: 在这里,25+25+13.125 = 63.125。我不太擅长使用模算术来即时生成这些数字,但我相信有一种好的方法可以从这些数字中推导出系列1,1,2.525 - kreativitea
@DavidZaslavsky 谢谢。我尝试了您的解决方案,但是出现了一个错误:TypeError: object of type 'itertools.chain' has no len()。不确定它的含义。Juniper,谢谢,很棒的解决方案,对于非常长的数据集有点慢,但它能工作。 - Dnaiel
@Dnaiel 哦,是的,我只是凭空写了这个代码,没有测试过(因此有“大概是这样”的说法)。出现错误是因为 itertools.chain 返回一个生成器,但 partition 需要一个列表,所以你可以像这样修复它:partition(list(itertools.chain(...)))。但它只是 juniper- 解决方案的简化版本,没有添加任何新内容。 - David Z
显示剩余2条评论

3

这是一个相当困难的问题,但我认为你正在把它变得比实际上更加复杂。我将从几个观察开始。

观察1. 您可以将许多因素分解出来,直到最后才减少编码量。而不是通过除以和乘以25(这很快就变得非常复杂),请将该操作保存到最后。

list2resized = [i/25.625 for i in [(5.6*25+7.8*0.625), 
                                   (7.8*24.375+8.9*(25.65-24.375)), 
                                   (23.725*8.9+(25.625-23.725)*13.4),...]]

# consider using ratios, rather than division
list2resized = [i * 1.025 for i in [(5.6 * 1 + 7.8 * .025), 
                                    (7.8 * .975 + 8.9 * .050), 
                                    (8.9 * .95 + 13.4 * .075),...]]

观察2.每个进行项的系数因此是25的递增步长。在之后保存除以1000 - 如果选择,您可以将整个方程乘以1000并使用模运算符...

 list2resized = [i * 1025/1000 for i in [(5.6 * 1000 + 7.8 * 25), # 1025 steps in
                                          (7.8 * 975 + 8.9 * 50), # 2050 steps in
                                          (8.9 * 950 + 13.4 * 75) # 3075 steps in

观察3.

在最终调整大小时,每个“bin”需要1.025的长度(假设有41个起始的bins,但最终取决于要调整的列表的长度)。1.0 * list[0] + .025 * list[1] 考虑到观察2,您可以将整个方程重写为一系列-

# the sum of the coefficients is always equal to the resize ratio
(1 * n1) + (.025 * n2)
(.975 * n2) + (.050 * n3) 
(.950 * n3) + (.075 * n4)
...

现在,您可以生成这些系数--
a = [i/40.0 for i in range(0, 40)][1:]
b = [1 - i/40.0 for i in range(0, 40)]

但这些情况都很容易处理,因为“旋转”永远不会重复。您只需在每个方程的各个部分中迭代每个箱中的系数,然后将它们压缩在一起并求和即可。这仅将列表压缩到其原始大小的最大一半。在这种情况下,您应该使用上述算法,它比您可以想象的任何其他算法都要快得多,因为它只是创建了一个数字列表,然后通过列表推导进行乘法。

但是,复杂的情况是当您有101个数字时,其中超过一个术语(有时甚至是第四个!)出现...

101/40.0 = 2.525 
# your bins need to be 2.525 units long.  

data = [4.1, 5.6, 10.3, 9.8, 40, 30, 21.4, 3, 2,...]

# calculated by hand
(1 * n1) + (1 * n2) + (.525 * n3) 
(.475 * n3) + (1 * n4) + (1 * n5) + (.05 * n6)
(.95 * n6) + (1 * n7) + (.575 * n8)
(.425 * n8) + (1 * n9) + (1 * n10) + (.100 * n11)

因此,我们需要一种更好的方法来生成系数。正如之前观察到的(3),最终项中系数的和是旧项目与新项目的比率。

101:40 = 2.525:1
41:40 = 1.025:1

接下来是生成系数。我们将使用一个嵌套列表的数据结构,它会迭代子列表直到没有任何内容为止。

[(1, 1, .525), (.475, 1, 1, .05) ...]

第一个子列表映射到新列表中的第1项。第二个子列表映射到第2项,以此类推,一直到末尾。所有子列表中所有项的总和应等于原始列表中的项n(在本例中为101)。
我现在要发布这篇文章,因为我必须实际工作。我会尽量回来继续努力。
以下是一个生成系数的函数。
n = 1000
d = 2525
items = 101
def coefficients(n, d, items):
    start = [n for i in xrange(items)]
    result = []
    p = []
    while True:
        while sum(p) < d:
            try:
                p.append(start.pop())
            except IndexError:
                return result
        extra = sum(p) % d
        p[-1] = n - extra
        result.append(p)
        p = [extra]

迭代系数以返回您的最终列表40。如果需要更多帮助,请告诉我。


非常感谢,这里有很棒的想法。如果您有时间,我仍然很想听听您的更多建议,但到目前为止,这是一个非常好的想法。这绝对是最快的实现方式,由于我的列表相当长,拥有一个快速选项会很不错。 - Dnaiel
我自己也在这个思路上考虑过,但是我没有时间去解决细节问题。不错 :-) - David Z
@Dnaiel,我添加了一个生成系数的函数,现在应该是完整的答案了。 - kreativitea

2

我对Python还比较新,因此您需要其他人来评估它的Python风格、优雅度和速度。

class StretchableList(list):
    def stretch(self, newlen):
        old = [ (i * (newlen-1), self[i]) for i in range(len(self)) ]
        new = [ i * (len(self)-1) for i in range(newlen) ]
        self[:] = []
        for n in new:
            while len(old) > 1 and n >= old[1][0]:
                old.pop(0)
            if old[0][0] == n:
                self.append(old[0][1])
            else:
                self.append( old[0][1] + \
                             float((n-old[0][0]))/(old[1][0]-old[0][0]) * \
                             (old[1][1]-old[0][1]) )
        return self

基本上,它定义了list的一个子类,只添加了一个名为stretch的方法。调用它并传入所需的新长度,它将会被拉伸或压缩到新的长度。我对加权平均值的计算方式与你略有不同...也许它等价,也可能不等价,但我假设数学部分可以根据需要进行修改。

@glidbud。感谢您的回答,我还没有测试过,但似乎是一个不错的答案。为了速度比较,我会将其与其他答案进行比较。 - Dnaiel

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接