"""真实pandas接口补充验收；需要pandas3.0.1，不是Vibe全流程。"""
import json
import platform
import pandas as pd


class SignalEngine:
    semantics = 'target_long_state'

    def generate(self, data_map):
        if not data_map:
            raise ValueError('empty_data_map')
        result = {}
        for symbol, frame in data_map.items():
            if not isinstance(frame, pd.DataFrame) or 'close' not in frame:
                raise ValueError('frame_or_close')
            if not frame.index.is_unique or not frame.index.is_monotonic_increasing:
                raise ValueError('input_index')
            if frame.empty or frame['close'].isna().any():
                raise ValueError('empty_or_missing_close')
            average = frame['close'].rolling(3, min_periods=3).mean()
            result[symbol] = (frame['close'] > average).astype('int64')
        return result


def validate(engine, data_map, output):
    if getattr(engine, 'semantics', None) != 'target_long_state':
        raise ValueError('semantics')
    if not isinstance(output, dict) or set(output) != set(data_map):
        raise ValueError('output_keys_or_type')
    for symbol, frame in data_map.items():
        if not frame.index.is_unique:
            raise ValueError('duplicate_index')
        series = output[symbol]
        if not isinstance(series, pd.Series):
            raise ValueError('not_series')
        if not series.index.is_unique or not series.index.equals(frame.index):
            raise ValueError('output_index')
        if series.isna().any() or not series.isin([0, 1]).all():
            raise ValueError('signal_domain')


dates = pd.date_range('2026-01-05', periods=3)
data = {'US:A': pd.DataFrame({'close': [10, 10, 12]}, index=dates),
        'US:B': pd.DataFrame({'close': [10, 10, 8]}, index=dates)}
saved = {s: frame.copy(deep=True) for s, frame in data.items()}
engine = SignalEngine()
output = engine.generate(data)
validate(engine, data, output)
assert output['US:A'].tolist() == [0, 0, 1]
assert output['US:B'].tolist() == [0, 0, 0]
for symbol in data:
    pd.testing.assert_frame_equal(saved[symbol], data[symbol])
duplicate = data['US:A'].copy()
duplicate.index = [dates[0], dates[0], dates[2]]
tests = [('wrong_return', data, []), ('wrong_keys', data, {'A': output['US:A']}),
         ('wrong_series', data, dict(output, **{'US:A': [0, 0, 1]})),
         ('duplicate_index', {'US:A': duplicate}, {'US:A': output['US:A']}),
         ('wrong_index', data, dict(output, **{'US:A': pd.Series([0, 0, 1])})),
         ('wrong_domain', data, dict(output, **{'US:A': pd.Series([0, 0, -1], index=dates)}))]
rejected = []
for label, inputs, outputs in tests:
    try:
        validate(engine, inputs, outputs)
    except ValueError:
        rejected.append(label)
assert len(rejected) == 6
engine.semantics = None
try:
    validate(engine, data, output)
except ValueError as error:
    assert str(error) == 'semantics'
else:
    raise AssertionError('未声明语义必须拒绝')
print(json.dumps({'python': platform.python_version(), 'pandas': pd.__version__,
                  'signals': {s: x.tolist() for s, x in output.items()},
                  'rejected': rejected, 'undeclared_semantics_rejected': True,
                  'input_unchanged': True, 'scope': 'actual pandas objects; no Vibe runner'}))