#!/usr/bin/env python
# -*- coding: utf-8 -*-
__all__ = ["ts_merge", "ts_splice"]
import pandas as pd
import numpy as np
import warnings
from vtools.functions.colname_align import align_inputs_strict
from vtools.data.indexing import resolve_common_freq, regular_index_from_valid_extent
[docs]
def _merge_output_index_irregular(series, *, strict_priority):
"""
Build the output index for ts_merge when frequency is not being preserved.
"""
if strict_priority:
pieces = []
for s in series:
lo = s.first_valid_index()
hi = s.last_valid_index()
if lo is None or hi is None:
continue
idx = s.index[(s.index >= lo) & (s.index <= hi)]
pieces.append(idx)
if not pieces:
return series[0].index[:0]
out = pieces[0]
for idx in pieces[1:]:
out = out.union(idx, sort=False)
return out.sort_values()
warnings.warn(
"ts_merge(..., preserve_freq=False, strict_priority=False) may produce "
"irregular union-index artifacts; this mode is not recommended.",
UserWarning,
stacklevel=2,
)
out = series[0].index
for s in series[1:]:
out = out.union(s.index, sort=False)
return out.sort_values()
[docs]
def _build_output_index(series, *, preserve_freq, fallback_builder):
"""
Decide the output index for a composition routine.
Parameters
----------
series : sequence of Series/DataFrame
preserve_freq : bool
If True, require agreement among all freq-bearing indexes and return
a regular target index spanning the global valid-data extent.
fallback_builder : callable
Called as fallback_builder(series) when preserve_freq does not apply.
Returns
-------
pandas.Index
"""
output_freq = resolve_common_freq(
[s.index for s in series],
preserve_freq=preserve_freq,
)
if output_freq is not None:
return regular_index_from_valid_extent(series, output_freq)
return fallback_builder(series)
#####################
[docs]
@align_inputs_strict(seq_arg=0, names_kw="names")
def ts_merge(series, names=None, strict_priority=False, preserve_freq=True):
"""
Merge multiple time series together, prioritizing series in order.
Parameters
----------
series : sequence of pandas.Series or pandas.DataFrame
Higher priority first. All indexes must be DatetimeIndex.
names : None, str, or iterable of str, optional
- If `None` (default), inputs must share compatible column names.
- If `str`, the output is univariate and will be named accordingly.
- If iterable, it is used as a subset/ordering of columns.
strict_priority : bool, default False
If False (default): lower-priority data may fill NaNs in higher-priority
series anywhere (traditional merge/overlay).
If True: for each column, within the window
[first_valid_index, last_valid_index] of any higher-priority series,
lower-priority data are masked out — even if the higher-priority value is NaN.
Outside those windows, behavior is unchanged.
preserve_freq : bool, default True
If True, attempt to preserve a common regular frequency across inputs.
When a common regular frequency can be established, the output is
reindexed to that regular extent. When it cannot be established,
ts_merge falls back to index-based merge behavior, which for strict_priority=True
is a patchwise combination of the indexes and for False is a shuffle.
Returns
-------
pandas.Series or pandas.DataFrame
"""
# --- Input validation (messages match tests) ---
if not isinstance(series, (list, tuple)) or len(series) == 0:
raise ValueError("`series` must be a non-empty tuple or list")
if not all(isinstance(getattr(s, "index", None), pd.DatetimeIndex) for s in series):
raise ValueError("All input series must have a DatetimeIndex.")
# If any DataFrame is present, convert Series->DataFrame to unify shapes
any_df = any(isinstance(s, pd.DataFrame) for s in series)
if any_df:
series = [
s.to_frame(name=s.name) if isinstance(s, pd.Series) else s for s in series
]
# Column compatibility checks (messages match tests)
all_df = all(isinstance(s, pd.DataFrame) for s in series)
any_df = any(isinstance(s, pd.DataFrame) for s in series)
any_series = any(isinstance(s, pd.Series) for s in series)
if all_df:
if names is None:
cols0 = list(series[0].columns)
for s in series[1:]:
if list(s.columns) != cols0:
raise ValueError(
"All input DataFrames must have the same columns when `names` is None."
)
elif any_df and any_series:
if names is None:
df_cols = {
c for s in series if isinstance(s, pd.DataFrame) for c in s.columns
}
for s in series:
if isinstance(s, pd.Series) and s.name not in df_cols:
raise ValueError(
"Mixed Series and DataFrames require Series names to match DataFrame columns."
)
# Build output index according to frequency / priority policy
original_all_series = all(isinstance(s, pd.Series) for s in series)
full_index = _build_output_index(
series,
preserve_freq=preserve_freq,
fallback_builder=lambda seq: _merge_output_index_irregular(
seq,
strict_priority=strict_priority,
),
)
# Align to union index and keep DataFrame shape for column-wise masking
aligned = []
for s in series:
a = s.reindex(full_index)
if isinstance(a, pd.Series):
a = a.to_frame(name=a.name)
aligned.append(a.copy())
# --- Strict priority dominance windows ---
if strict_priority:
# For each column, mask lower-priority data only within the dominance window
for col in set(c for a in aligned for c in a.columns):
# Find the highest-priority series that contains this column
for i, hi in enumerate(aligned):
if col in hi.columns:
s_col = hi[col]
lo_idx = s_col.first_valid_index()
hi_idx = s_col.last_valid_index()
break
else:
continue # column not present in any input
if lo_idx is None or hi_idx is None:
continue # no dominance window for this column
mask = (full_index >= lo_idx) & (full_index <= hi_idx)
for j in range(i + 1, len(aligned)):
if col in aligned[j].columns:
aligned[j].loc[mask, col] = np.nan # keep numeric dtype
# Combine in priority order via combine_first
merged = aligned[0]
for a in aligned[1:]:
merged = merged.combine_first(a)
# NEW: ensure the full union index is present (including all-NaN rows)
merged = merged.reindex(full_index)
# --- keep union, but drop masked lower-only rows inside dominance window ---
if strict_priority:
# Use the highest-priority input to define the dominance window (per-column union)
top = aligned[0] # DataFrame (we normalized earlier)
lo_candidates = [top[c].first_valid_index() for c in top.columns]
hi_candidates = [top[c].last_valid_index() for c in top.columns]
lo0 = min([x for x in lo_candidates if x is not None], default=None)
hi0 = max([x for x in hi_candidates if x is not None], default=None)
if lo0 is not None and hi0 is not None:
idx = merged.index
in_window = (idx >= lo0) & (idx <= hi0)
in_top_idx = idx.isin(
series[0].index
) # keep rows that are from the top series' index
# rows that are entirely NaN after strict masking
all_nan = (
merged.isna().all(axis=1)
if isinstance(merged, pd.DataFrame)
else merged.isna()
)
# Drop only those that are NaN, within the window, and not in the top index (i.e., lower-only timestamps)
drop_mask = in_window & (~in_top_idx) & all_nan
if drop_mask.any():
merged = merged.loc[~drop_mask]
# If all inputs were univariate Series, return a Series
if original_all_series:
merged = merged.squeeze()
else:
if isinstance(merged, pd.Series):
merged = merged.to_frame()
return merged
[docs]
@align_inputs_strict(seq_arg=0, names_kw="names")
def ts_splice(
series, names=None, transition="prefer_last", floor_dates=False, preserve_freq=True
):
"""
Splice multiple time series together, prioritizing series in patches of time.
Unlike `ts_merge`, which blends overlapping data points, `ts_splice` stitches
together time series without overlap. The function determines when to switch
between series based on a transition strategy.
Parameters
----------
series : tuple or list of pandas.DataFrame or pandas.Series
A tuple or list of time series.
Each series must have a `DatetimeIndex` and consistent column structure.
names : None, str, or iterable of str, optional
- If `None` (default), all input series must share common column names,
and the output will merge common columns.
- If a `str`, all input series must have a **single column**, and the
output will be a DataFrame with this name as the column name.
- If an iterable of `str`, all input DataFrames must have the same
number of columns matching the length of `names`, and these will be
used for the output.
transition : {'prefer_first', 'prefer_last'} or list of pandas.Timestamp
Defines how to determine breakpoints between time series:
- `'prefer_first'`: Uses the earlier series on the list during until its valid timestamp.
- `'prefer_last'`: Uses the later series starting from its first valid timestamp.
- A list of specific timestamps can also be provided as transition points.
floor_dates : bool, optional, default=False
If `True`, inferred transition timestamps (`prefer_first` or `prefer_last`)
are floored to the beginning of the day. This can introduce NaNs if the
input series are regular with a `freq` attribute.
preserve_freq : bool, default True
If True, attempt to preserve a common regular frequency across inputs.
When a common regular frequency can be established, the output is
reindexed to that regular extent. When it cannot be established,
ts_merge falls back to index-based merge behavior, which for strict_priority=True
is a patchwise combination of the indexes and for False is a shuffle.
Returns
-------
pandas.DataFrame or pandas.Series
- If the input contains multi-column DataFrames, the output is a DataFrame
with the same column structure.
- If a collection of single-column `Series` is provided, the output will be
a Series.
- The output retains a `freq` attribute if all inputs share the same frequency.
Notes
-----
- The output time index is the union of input time indices.
- If `transition` is `'prefer_first'`, gaps may appear in the final time series.
- If `transition` is `'prefer_last'`, overlapping data is resolved in favor
of later series.
See Also
--------
ts_merge : Merges series by filling gaps in order of priority.
"""
series = [s.copy() for s in series]
if not isinstance(series, (tuple, list)) or len(series) == 0:
raise ValueError(
"`series` must be a non-empty tuple or list of pandas.Series or pandas.DataFrame."
)
if not all(
isinstance(s.index, pd.DatetimeIndex) or isinstance(s.index, pd.PeriodIndex)
for s in series
):
raise ValueError("All input series must have a DatetimeIndex or PeriodIndex.")
# Ensure all input series have the same type of index.
index_type = type(series[0].index)
if not all(isinstance(s.index, index_type) for s in series):
raise ValueError(
f"All input series must have indexes of type {index_type.__name__}."
)
if transition not in ["prefer_last", "prefer_first"] and not isinstance(
transition, list
):
raise ValueError(
"`transition` must be 'prefer_last', 'prefer_first', or a list of timestamps."
)
# Decide whether a regular output index should be imposed at the end.
target_index = _build_output_index(
series,
preserve_freq=preserve_freq,
fallback_builder=lambda seq: None,
)
# If names is a string, pre-rename each series for consistency.
if names is not None and isinstance(names, str):
series = [
(
s.rename(columns={s.columns[0]: names})
if isinstance(s, pd.DataFrame)
else s.rename(names)
)
for s in series
]
# Compute transition points.
if isinstance(transition, list):
transition_points = transition
duplicate_keep = "first"
else:
if transition == "prefer_first":
transition_points = [s.last_valid_index() for s in series[:-1]]
duplicate_keep = "first"
elif transition == "prefer_last":
transition_points = [s.first_valid_index() for s in series[1:]]
duplicate_keep = "last"
if floor_dates:
transition_points = [dt.floor("D") for dt in transition_points]
transition_points = [None] + transition_points + [None]
# Extract sections based on transition points.
sections = []
for ts_obj, start, end in zip(
series, transition_points[:-1], transition_points[1:]
):
section = ts_obj.loc[start:end]
if not section.empty:
sections.append(section)
spliced = pd.concat(sections, axis=0).sort_index() if sections else pd.DataFrame()
# If all inputs were univariate, ensure output remains univariate.
univariate = all(
(s.name is not None if isinstance(s, pd.Series) else s.shape[1] == 1)
for s in series
)
if univariate and isinstance(spliced, pd.DataFrame):
spliced = spliced.iloc[:, 0]
# Normalize output type.
if all(isinstance(s, pd.Series) for s in series):
spliced = spliced.squeeze()
else:
if isinstance(spliced, pd.Series):
spliced = spliced.to_frame()
# Remove duplicate timestamps based on transition preference.
dup = spliced.index.duplicated(keep=duplicate_keep)
if dup.any():
spliced = spliced[~dup]
if target_index is not None:
spliced = spliced.reindex(target_index)
return spliced
