P5347. 时序数据修复与加权特征评分
场次:2026年9月4日 · 来源:CodeFun2000 P5347
题目描述
某数据采集系统按时间顺序获取了 条记录,每条记录包含时间戳 、数值 和质量标记 。
其中,质量标记用于表示数据状态:当 为 时表示数据有效,当 为 时表示数据存在可疑情况,当 为 时表示数据无效。系统需要按照无效数据修复、局部异常处理、标准化以及评分计算的顺序依次完成数据处理流程。
阶段一:无效数据修复
对每个满足 的位置 ,分别寻找其左侧和右侧最近的满足 的有效数据下标,记为 和 。如果不存在,则对应下标视为未找到。
- 若 和 均存在,则按时间戳做线性插值: 。
- 若只存在 ,则 。
- 若只存在 ,则 。
- 若 和 均不存在,则 。
其余非无效数据的值保持不变,即 。完成本阶段后得到序列 。
阶段二:局部异常处理
设 为奇数,令半径 。对每个位置 ,定义窗口下标范围 ,其中 ,。边界处不补值,因此实际窗口大小允许小于 。
对位置 ,基于该窗口内第一阶段结果 计算:
- 窗口均值 ;
- 窗口标准差 。
如果 ,则将当前值替换为窗口均值,即 ;否则保持 。本阶段所有窗口统计量均基于阶段一结束后的序列 ,不随本阶段的替换而改变。
阶段三:z-score 标准化
对阶段二得到的序列 的全部 个值,计算全局均值 和全局标准差 :
每个位置的标准化值为 。特别地,如果 ,则最终分数直接定义为 0。
阶段四:最终评分
根据质量标记定义质量权重 :
- 若 ,则 ;
- 若 ,则 ;
- 若 ,则 。
取第二阶段结果 的绝对值,定义原始值权重:
如果分母为 0,则最终分数直接定义为 0。
综合得到特征分数:
最终输出将 乘以 100 后四舍五入得到的整数。
约束条件
- 记录条数 满足 。
- 窗口大小 满足 ,且 为奇数。
- 标准差倍数 满足 。
- 数值 满足 。
- 时间戳 满足 ,且严格递增。
- 质量标记 只取
0、1、2中的一种。 - 至少存在一个 的有效数据记录。
输入格式
第一行包含三个整数 、、,分别表示记录数量、移动窗口大小和异常判断使用的标准差倍数。
接下来包含 行,其中第 行包含三个整数 、、,依次表示第 条记录的时间戳、数值和质量标记。
输出格式
输出一个整数,表示特征分数 乘以 100 并四舍五入后的结果。
示例
示例 1
输入:
1 1 1
1 5 0
输出:
0
说明:
只有 1 条记录,窗口大小 。无效数据修复阶段没有无效记录,因此 。
局部异常处理时窗口只包含自身,均值 ,标准差 。由于 ,不满足替换条件,所以 。
第三阶段全局标准差 ,根据规则最终分数直接定义为 0。
示例 2
输入:
3 1 1
1 1 0
2 2 1
3 3 2
输出:
102
说明:
第一阶段,第 3 条记录 。它左侧最近的 记录是第 1 条,右侧不存在,因此 。其余位置保持不变,得到 。
第二阶段 ,每个窗口只包含当前位置,窗口均值等于当前值,标准差为 0,因此所有位置均保持原值,。
第三阶段全局均值 ,全局标准差 。
第四阶段质量权重为 ,,。计算分母 。三个位置的 分别为 、、。加权求和得到 ,乘以 100 并四舍五入为 102。
示例 3
输入:
3 3 1
1 0 0
2 0 0
3 1 0
输出:
141
说明:
三条记录均为有效数据,因此无需无效数据修复,。
窗口大小 ,每个位置的窗口都覆盖全部 3 条记录,窗口均值 ,标准差 。位置 1 和位置 2 的偏差绝对值均为 ,不大于 ,保持原值。位置 3 的偏差绝对值为 ,大于 ,因此 。得到 。
第三阶段全局均值 ,全局标准差 。
第四阶段所有 ,所以 。分母 ,只有第 3 条记录的权重非零。其 ,因此 ,乘以 100 并四舍五入为 141。
解题思路
第一步:理解问题本质
这道题不需要想算法,需要严格按公式执行四个阶段。真正容易出错的地方不在“难”,而在“每个阶段到底用哪条序列”。
把数据处理链路画出来:
| 阶段 | 输入 | 输出 | 关键规则 |
|---|---|---|---|
| 一、无效数据修复 | 原始 | 序列 | 只处理 ,用最近的 插值 |
| 二、局部异常处理 | 序列 | 序列 | 窗口统计量全部来自 ,同时写入 |
| 三、z-score 标准化 | 序列 | 用 的全局均值与全局标准差 | |
| 四、加权评分 | 、、原始 | 一个整数 | 权重用 ,绝对值用 |
三条“不能混”的边界:
- 阶段二的窗口统计量基于 ,不是边算边替换之后的 ;
- 阶段三的标准差是全局的(对整条序列),不是窗口的;
- 阶段四的质量权重 来自原始质量标记 ,而不是修复后的状态。
另外两个容易忽略的约定:
- 标准差一律是总体标准差:除以元素个数 (或 ),不是除以 ;
- 边界处的窗口不补值,实际窗口大小可以小于 。
第二步:暴力解法——按定义逐位置重扫窗口
最直白的写法是:每个位置都把窗口里的数重新取出来,算均值和标准差。
import math
half = (w - 1) // 2
b = [0.0] * n
for i in range(n):
lo = max(0, i - half)
hi = min(n - 1, i + half)
window = a[lo:hi + 1]
m = len(window)
mu = sum(window) / m
var = sum((x - mu) ** 2 for x in window) / m
sigma = math.sqrt(var)
b[i] = mu if abs(a[i] - mu) > k * sigma else a[i]
阶段一如果也照定义做——对每个 的位置分别向左、向右扫到第一个 ——那一步是 ;阶段二是 。本题 、,最坏情况下 次浮点运算,Python 会超时。
两处都能优化掉:阶段一左右各扫一遍就能预处理出每个位置的“最近有效点”,阶段二用前缀和在 内拿到窗口的和与平方和。
第三步:阶段一与阶段二降到线性
阶段一:预处理最近的有效点。
从左往右扫一遍,维护“上一个 的位置”;每个位置先记录当前值,再看自己是不是 ,这样得到的自然是严格左侧最近的有效点。右侧同理,从右往左扫一遍。
有了这两张表,每个 的位置都能 拿到插值端点:
注意分母是时间戳之差 ,不是下标之差——题目按时间插值,时间戳严格递增所以不会除零。
阶段二:前缀和求窗口统计量。
令 、,那么窗口 内
于是
这里有一个数值上的坑:如果数值本身很大(例如都接近 )而彼此只差一点点, 会把有效数字挤掉,相减得到的方差就不可信。所以先把整条序列平移一个参考值(代码里取 )再算前缀和,平方和只涉及“差异”,量级小得多。
还有两处保护:
- 浮点相减可能得到极小的负数,开方前夹回 ;
- 比较条件是严格大于 ,等于阈值时不替换。
第四步:阶段三、四与收尾细节
阶段三:对 求全局均值和总体标准差。若标准差为 (等价于 中所有值相同),题目规定最终分数直接为 。
判断“全相等”时用 max(b) == min(b),而不是判断算出来的方差是否等于 :均值本身是浮点累加的结果,可能存在 量级的抖动,用极值比较既精确又直观。
阶段四:质量权重按原始标记取 、、,然后
分母为 时(所有 都是 )同样直接输出 。
四舍五入:题目要求四舍五入,而 Python 的 round() 是“四舍六入五成双”(银行家舍入),round(0.5) 得到 0。所以用
int(math.floor(S * 100 + 0.5 + 1e-9))
末尾的 是为了让“理论值恰好是 ”的情况不被浮点误差压到下半区。
累加精度:阶段三、四都是对整条序列求和,用 math.fsum 做精确求和,比 sum 更稳。
完整代码实现
"""
P5347. 时序数据修复与加权特征评分 - 四阶段按公式模拟
输入格式:
- 第 1 行:三个整数 n、w、k(w 为奇数);
- 接下来 n 行:每行三个整数 t_i、v_i、q_i。
输出格式:
- 一个整数,即 S 乘以 100 后四舍五入的结果。
时间复杂度:O(n)
空间复杂度:O(n)
"""
import math
import sys
def solve():
data = sys.stdin.buffer.read().split()
if not data:
return
it = iter(data)
n = int(next(it))
window = int(next(it))
k = float(next(it))
times = [0] * n
values = [0] * n
quality = [0] * n
for i in range(n):
times[i] = int(next(it))
values[i] = int(next(it))
quality[i] = int(next(it))
# ---------------- 阶段一:无效数据修复 ----------------
# 左边最近的 q=0:先记录再看自己,保证是“严格左侧”
left_valid = [-1] * n
last_valid = -1
for i in range(n):
left_valid[i] = last_valid
if quality[i] == 0:
last_valid = i
right_valid = [-1] * n
next_valid = -1
for i in range(n - 1, -1, -1):
right_valid[i] = next_valid
if quality[i] == 0:
next_valid = i
a = [0.0] * n
for i in range(n):
if quality[i] != 2:
a[i] = float(values[i]) # 有效数据和可疑数据都原样保留
continue
left = left_valid[i]
right = right_valid[i]
if left >= 0 and right >= 0:
# 两端都有:按时间戳线性插值
v_left = float(values[left])
v_right = float(values[right])
a[i] = v_left + (v_right - v_left) * (times[i] - times[left]) / (
times[right] - times[left]
)
elif left >= 0:
a[i] = float(values[left])
elif right >= 0:
a[i] = float(values[right])
else:
a[i] = float(values[i])
# ---------------- 阶段二:局部异常处理 ----------------
half = (window - 1) // 2
# 先整体平移 a[0],再算前缀和:避免“数值很大、差异很小”时
# 平方和相减丢失有效数字
reference = a[0]
diff = [x - reference for x in a]
prefix = [0.0] * (n + 1)
prefix_square = [0.0] * (n + 1)
for i in range(n):
d = diff[i]
prefix[i + 1] = prefix[i] + d
prefix_square[i + 1] = prefix_square[i] + d * d
b = [0.0] * n
for i in range(n):
# 边界处不补值:窗口下标直接截断
low = i - half
if low < 0:
low = 0
high = i + half
if high > n - 1:
high = n - 1
count = high - low + 1
sum_d = prefix[high + 1] - prefix[low]
sum_square = prefix_square[high + 1] - prefix_square[low]
mean_d = sum_d / count
variance = sum_square / count - mean_d * mean_d
if variance < 0.0:
variance = 0.0 # 浮点相减可能出现极小负数
sigma = math.sqrt(variance)
mean = reference + mean_d
# 严格大于才替换;统计量全部来自 a,不随替换改变
if abs(a[i] - mean) > k * sigma:
b[i] = mean
else:
b[i] = a[i]
# ---------------- 阶段三:z-score 标准化 ----------------
# 全相等 <=> 全局标准差为 0,用极值比较更稳
if max(b) == min(b):
print(0)
return
mean = math.fsum(b) / n
variance = math.fsum((x - mean) * (x - mean) for x in b) / n
std = math.sqrt(variance)
# ---------------- 阶段四:加权评分 ----------------
weights = []
for q in quality:
if q == 0:
weights.append(1.0)
elif q == 1:
weights.append(0.5)
else:
weights.append(0.25)
denominator = math.fsum(p * abs(x) for p, x in zip(weights, b))
if denominator == 0.0:
print(0)
return
score = math.fsum(
(p * abs(x) / denominator) * (abs(x - mean) / std)
for p, x in zip(weights, b)
)
# 四舍五入:round() 是银行家舍入,这里用 floor(x + 0.5)
print(int(math.floor(score * 100 + 0.5 + 1e-9)))
if __name__ == "__main__":
solve()
NumPy 版本
以下版本使用 NumPy。只有在评测环境允许第三方库时才能提交;CodeFun2000 的普通 Python 环境不一定安装 NumPy。
思路与纯 Python 版完全一致,四个阶段都改成对整条序列的运算:阶段一用 np.searchsorted 在有效点下标表上一次定位左右端点;阶段二用 np.cumsum 的前缀和、前缀平方和取窗口统计量;阶段三、四都是整数组的广播运算。
"""
P5347. 时序数据修复与加权特征评分 - NumPy 版
四个阶段全部改成对整条序列的向量化运算:
- 阶段一用 np.searchsorted 在“有效点下标表”上一次定位左右端点,再按三种情形取值;
- 阶段二用 np.cumsum 求前缀和/前缀平方和,窗口均值与标准差由切片相减直接得到;
- 阶段三、四都是整数组的广播运算。
时间复杂度:O(n);空间复杂度:O(n)。
"""
import sys
import numpy as np
def solve():
data = sys.stdin.buffer.read().split()
if not data:
return
n = int(data[0])
window = int(data[1])
k = float(data[2])
table = np.array(list(map(int, data[3:3 + 3 * n])), dtype=np.int64).reshape(n, 3)
times = table[:, 0].astype(np.float64) # 时间戳:阶段一按它插值
values = table[:, 1].astype(np.float64)
quality = table[:, 2]
# ---------------- 阶段一:无效数据修复 ----------------
# 有效点(q=0)的下标表;对每个 q=2 的位置二分,得到“右侧最近”和“左侧最近”
valid = np.flatnonzero(quality == 0)
broken = np.flatnonzero(quality == 2)
a = values.copy()
if broken.size and valid.size:
pos = np.searchsorted(valid, broken, side="left") # 第一个 >= 自己的有效点
has_left = pos > 0
has_right = pos < valid.size
# q=2 的位置本身不在有效点表里,所以 valid[pos] 严格在右侧、valid[pos-1] 严格在左侧
left = valid[np.maximum(pos - 1, 0)]
right = valid[np.minimum(pos, valid.size - 1)]
v_left = values[left]
v_right = values[right]
both = has_left & has_right
# 两端都有:按时间戳线性插值(分母是时间戳之差,不是下标之差)。
# 单边存在时显式把分母置 1,避免算出 0 除的 inf/nan(那些位置随后会被丢弃)。
span = np.where(both, times[right] - times[left], 1.0)
interp = v_left + (v_right - v_left) * (times[broken] - times[left]) / span
a[broken] = np.where(
both,
interp,
np.where(has_left, v_left, np.where(has_right, v_right, values[broken])),
)
# ---------------- 阶段二:局部异常处理 ----------------
half = (window - 1) // 2
# 先整体平移 a[0] 再算前缀和:数值都很大、彼此差异很小时,平方和相减才不会丢有效数字
reference = a[0]
diff = a - reference
prefix = np.zeros(n + 1, dtype=np.float64)
np.cumsum(diff, out=prefix[1:])
prefix_square = np.zeros(n + 1, dtype=np.float64)
np.cumsum(diff * diff, out=prefix_square[1:])
index = np.arange(n)
low = np.maximum(index - half, 0) # 边界处不补值,窗口直接截断
high = np.minimum(index + half, n - 1)
count = (high - low + 1).astype(np.float64)
sum_diff = prefix[high + 1] - prefix[low]
sum_square = prefix_square[high + 1] - prefix_square[low]
mean_diff = sum_diff / count
variance = np.maximum(sum_square / count - mean_diff * mean_diff, 0.0)
sigma = np.sqrt(variance) # 总体标准差,除以元素个数
mean = reference + mean_diff
# 严格大于才替换;统计量全部来自 a,不随本阶段的替换而改变
b = np.where(np.abs(a - mean) > k * sigma, mean, a)
# ---------------- 阶段三:z-score 标准化 ----------------
if b.max() == b.min(): # 全相等 <=> 全局标准差为 0
print(0)
return
mean_b = b.mean()
std_b = b.std() # numpy 的 std 默认 ddof=0,即总体标准差
# 说明:numpy 的 mean/std/sum 用成对求和,与 math.fsum 的差异在 1e-16 量级,
# 只要输入不是恰好落在 x.5 的舍入边界上,最终整数完全一致。
# ---------------- 阶段四:加权评分 ----------------
# 质量权重来自原始质量标记:q=0 -> 1,q=1 -> 0.5,q=2 -> 0.25
weights = np.where(quality == 0, 1.0, np.where(quality == 1, 0.5, 0.25))
numerator = weights * np.abs(b)
denominator = numerator.sum()
if denominator == 0.0:
print(0)
return
# 逐项 (p_i|b_i|/分母) * |z_i| 再求和,与公式逐字对应
score = (numerator / denominator * (np.abs(b - mean_b) / std_b)).sum()
# 四舍五入:floor(x + 0.5),末尾 1e-9 防止 x.5 被浮点误差压到下半区
print(int(np.floor(score * 100 + 0.5 + 1e-9)))
if __name__ == "__main__":
solve()
示例推演
示例 1
、、,唯一记录是 。
- 阶段一:没有 的记录,;
- 阶段二:,窗口只有自己,、, 不满足严格大于,;
- 阶段三:只有一个值,全局标准差 ,按规则直接输出 。
0
示例 2
、、,记录为 、、。
阶段一:第 3 条 ,它左边最近的 是第 1 条,右边没有,所以 。第 2 条是 (可疑但保留原值),所以
阶段二: 所以 ,每个窗口只有自己:、,偏差为 不满足严格大于,全部保持原值:
阶段三:
三个 分别是
阶段四:,所以 ,,
三个权重分别是 、、,于是
,四舍五入为 102。
示例 3
、、,三条记录都是 ,值为 ,所以 。
阶段二:,每个位置的窗口按 截断(下标从 1 起):
| 位置 | 窗口 | 窗口均值 | 标准差 | 是否 | ||
|---|---|---|---|---|---|---|
| 1 | 否 | |||||
| 2 | 否 | |||||
| 3 | 否(严格大于才替换) |
所以 。
阶段三:
,。
阶段四:所有 ,所以 ;。只有第 3 条的 非零,因此它的权重是 ,另外两条是 :
,四舍五入为 141。
说明:题面的示例说明把三个窗口都描述成覆盖全部记录;按形式化定义(“边界处不补值”),位置 1 和位置 3 的窗口会截断成 2 个元素。两种口径都得到 :因为截断口径下的 与整窗口径下的 只差一个全局缩放系数,而 和权重都对统一的缩放不敏感。实现以形式化定义为准。
复杂度分析
| 解法 | 时间复杂度 | 空间复杂度 | 说明 |
|---|---|---|---|
| 完全按定义(阶段一逐点扫描 + 阶段二逐位置重扫窗口) | 时约 次运算,Python 会超时 | ||
| 阶段一预处理(左右各扫一遍) | 两个辅助数组记录最近有效点 | ||
| 预处理 + 前缀和(本题) | 窗口均值、方差都是 查询 |
,线性做法的运算量在 量级,瓶颈反而在输入解析和浮点运算本身。
易错点总结
1. 阶段二边算边改
题目说得很清楚:本阶段所有窗口统计量都基于阶段一结束后的 。如果一边替换 一边算后面的窗口,后面的窗口就会混入被替换过的值,误差会沿着序列一路传播。正确做法是读 、写 ,两个数组分开。
2. 边界窗口没有截断
、,所以两端的窗口会变窄,窗口大小小于 。若补零或补边界值凑满 ,均值与标准差都会被拉偏。
3. 把标准差算成样本标准差
题目公式里分母是元素个数( 和 ),是总体标准差。写成 或 (样本标准差)会让阈值和 都发生偏移。
4. 阶段一找的是“最近的有效点”,不是“最近的记录”
的记录是可疑数据,不能用来插值;只有 才是插值端点。同时 自身的值保持不变(),它既不参与修复别人,也不被别人修复。
5. 插值用下标差除
、、 是时间戳,不是下标。时间戳可能间隔很不均匀(例如 ),按下标插值会得到完全不同的结果。分母是 。
6. 阶段四的权重与绝对值取错来源
- 质量权重 来自原始 ,不是修复后的状态;
- 里用的是阶段二结果 的绝对值,不是 ,也不是 。
7. 忘记两条“直接输出 0”
- 全局标准差 :所有 相同;
- 分母 :所有 都是 。
这两种情况都要在除法之前短路,否则会除零或得到 nan。
8. 用 round() 做四舍五入
Python 的 round 采用银行家舍入,round(2.5)=2、round(3.5)=4。题目要的是通常意义的四舍五入,应使用 floor(x + 0.5)。
9. 方差用 时不做保护
当前缀和相减得到极小的负数时,math.sqrt 会抛 ValueError: math domain error。开方前夹回 ;数值量级很大时先把序列平移一个参考值,减少有效数字损失。
10. 判断“标准差为 0”用浮点等值
直接写 if s == 0 依赖均值累加的精确性。用 max(b) == min(b) 判断“所有值相同”更可靠,也正好等价于数学上的 。
扩展思考
- 更稳健的异常判定:均值和标准差本身会被异常值拉偏(掩蔽效应)。工程上常用中位数与 MAD(中位数绝对偏差)代替,或者用滑动中位数,判断逻辑仍然是“偏差超过 倍尺度”。
- 流式数据:如果记录是一条条到达的,阶段二的窗口可以用双端队列 + 增量维护和与平方和,做到 摊还;但要注意增量更新会累积浮点误差,必要时定期重算。
- 统一缩放不变性: 对 的平移和缩放都不敏感,权重 对统一缩放也不敏感。所以真正影响 的只有 的相对形状。这解释了示例 3 中两种窗口口径为什么给出同一个答案。
- 精确算术:本题的输入都是整数,理论上插值后的值是有理数,可以用
fractions.Fraction精确计算,再用decimal做最后的舍入。代价是速度慢很多,只适合对边界舍入极度敏感的场合。 - 多通道时序:如果每条记录带多个数值通道,四个阶段都要对每个通道分别执行,最后再把各通道的分数加权合并。