ETF 申赎与两融示例

"""单只股票ETF：准备实物成分券后申购赎回，或买入ETF后赎回。

从空仓账户启动，运行一轮。现金及成分券均使用任务的同一个证券账户。
日线1d、分钟1M均可；初始资金需覆盖整篮资产、现金替代和相关费用。
示例处理沪深股票实物篮子，不处理境外或虚拟现金项目。
赎回后保留到账成分券以及买入取整产生的余股，不自动卖出。
实际交收数据未接通时保留待交收；不会自行填充清算结果。
"""

from datetime import datetime, timedelta
from math import ceil, floor, isfinite

from pandas import isna
import panda_data

from panda_backtest.api.api import *


STRATEGY_PARAMS = {
    'etf_symbol': '510050.SH',
    'basket_count': 1,
    'cash_buffer': 100000.0,
    'cash_substitution': False,
    'cash_symbols': [],
    'start_mode': '先申购再赎回',
}


def initialize(context):
    context.account = context.run_info.stock_account
    context.etf = STRATEGY_PARAMS['etf_symbol']
    context.baskets = STRATEGY_PARAMS['basket_count']
    context.cash_buffer = float(STRATEGY_PARAMS['cash_buffer'])
    context.cash_substitution = STRATEGY_PARAMS['cash_substitution']
    if type(context.baskets) is not int or context.baskets < 1:
        raise ValueError('申购篮数必须是正整数')
    if type(context.cash_substitution) is not bool:
        raise ValueError('本示例cash_substitution只接受True或False')
    cash_symbols = list(STRATEGY_PARAMS['cash_symbols'])
    if cash_symbols:
        # 指定代码列表时，仅主动替代这些成分券；必须替代项仍按清单执行。
        context.cash_substitution = {symbol: True for symbol in cash_symbols}
    if not isfinite(context.cash_buffer) or context.cash_buffer < 0:
        raise ValueError('现金预留必须是有限非负数')
    if len(context.etf) != 9 or context.etf[-3:] not in ('.SH', '.SZ'):
        raise ValueError('ETF请填写完整的沪深证券代码')
    account = _account(context)
    for position in account.positions.values():
        if position.quantity > 0:
            raise ValueError('一级篮子示例需要从空仓账户启动')

    mode = STRATEGY_PARAMS['start_mode']
    if mode not in ('先申购再赎回', '买入ETF后赎回'):
        raise ValueError('start_mode请使用先申购再赎回或买入ETF后赎回')
    context.phase = '准备成分券' if mode == '先申购再赎回' else '买入ETF'
    context.current_date = None
    context.attempted = set()
    context.wait_messages = set()
    context.pending = {}
    context.finished = set()
    context.seen_trades = set()
    context.component_symbols = set()
    context.last_buy_date = None
    context.primary_action = None
    context.primary_key = None
    context.primary_finished = set()
    context.primary_attempt_date = None
    context.created_units = 0
    context.created_date = None

    # 在初始化阶段加载全周期，分段请求以满足接口单次查询范围限制。
    # 沿用任务的数据认证，不调用init_token。
    context.pcf_by_date = {}
    context.limits_by_date = {}
    context.components_by_date = {}
    start = datetime.strptime(str(context.run_info.start_date), '%Y%m%d')
    end = datetime.strptime(str(context.run_info.end_date), '%Y%m%d')
    if start > end:
        raise ValueError('回测开始日期不能晚于结束日期')
    while start <= end:
        chunk_end = min(start + timedelta(days=179), end)
        params = {
            'symbol': context.etf, 'exchange': context.etf[-2:],
            'start_date': start.strftime('%Y%m%d'),
            'end_date': chunk_end.strftime('%Y%m%d'),
        }
        _merge(context.pcf_by_date, panda_data.get_fund_etf_cr(**params), context.etf, params)
        _merge(context.limits_by_date, panda_data.get_fund_etf_cr_limits(**params), context.etf, params)
        _merge(context.components_by_date, panda_data.get_fund_etf_constituents(**params), context.etf, params)
        start = chunk_end + timedelta(days=1)
    if not context.pcf_by_date or not context.limits_by_date or not context.components_by_date:
        raise ValueError('回测区间内的PCF、申赎额度或成分券数据为空')
    print(f'[初始化] 账户={context.account}，ETF={context.etf}，篮数={context.baskets}')


def _account(context):
    account = context.stock_account_dict.get(context.account)
    if account is None:
        raise ValueError(f'证券账户不存在：{context.account}')
    return account


def _value(row, name):
    value = row.get(name)
    return None if value is None or bool(isna(value)) else value


def _positive(value, name):
    if value is None or isinstance(value, bool):
        raise ValueError(f'{name}缺失或无效')
    number = float(value)
    if not isfinite(number) or number <= 0:
        raise ValueError(f'{name}必须是有限正数')
    return number


def _integer(value, name):
    number = _positive(value, name)
    if not number.is_integer():
        raise ValueError(f'{name}必须是整数')
    return int(number)


def _group(frame, symbol):
    grouped = {}
    if frame is None or frame.empty:
        return grouped
    for row in frame.to_dict('records'):
        if str(row.get('symbol')) != symbol:
            raise ValueError('业务数据返回了不匹配的ETF代码')
        date = str(row.get('date', '')).replace('-', '')[:8]
        if len(date) != 8 or not date.isdigit():
            raise ValueError('业务数据交易日期无效')
        grouped.setdefault(date, []).append(row)
    return grouped


def _merge(target, frame, symbol, params):
    if frame is None:
        raise ValueError('业务数据接口没有返回结果，不能视为已确认的空数据')
    for date, rows in _group(frame, symbol).items():
        if not params['start_date'] <= date <= params['end_date']:
            raise ValueError('业务数据返回了请求范围以外的日期')
        target.setdefault(date, []).extend(rows)


def _wait(context, reason):
    if reason not in context.wait_messages:
        context.wait_messages.add(reason)
        print(f'[{context.trade_date} {context.hms}] [等待] {reason}')


def _snapshot(context):
    today = str(context.trade_date)
    rules = context.pcf_by_date.get(today, [])
    limits = context.limits_by_date.get(today, [])
    components = context.components_by_date.get(today, [])
    if not rules or not limits or not components:
        _wait(context, '当日PCF、额度或成分券数据不齐')
        return None
    if len(rules) != 1 or len(limits) != 1:
        raise ValueError('同一交易日的PCF或额度数据存在重复行')
    rule, limit = rules[0], limits[0]
    rule_unit = _value(rule, 'unit')
    if rule_unit is None:
        rule_unit = _value(rule, 'min_cr_unit')
    limit_unit = _value(limit, 'min_cr_unit')
    if limit_unit is None:
        limit_unit = _value(limit, 'unit')
    if rule_unit is not None:
        rule_unit = _integer(rule_unit, 'PCF最小申赎单位')
    if limit_unit is not None:
        limit_unit = _integer(limit_unit, '额度表最小申赎单位')
    if rule_unit is not None and limit_unit is not None and rule_unit != limit_unit:
        raise ValueError('当日PCF和额度表的最小申赎单位不一致')
    unit = limit_unit if limit_unit is not None else rule_unit
    unit = _integer(unit, '最小申赎单位')
    nav = _value(rule, 'unit_nav')
    if nav is None:
        nav = _value(limit, 'nav')
    if nav is None:
        nav = _positive(_value(rule, 'creation_unit'), '整篮参考资产净值') / unit
    nav = _positive(nav, 'PCF参考单位净值')
    return rule, unit, nav, components


def _physical_targets(context, rows):
    targets, seen = {}, set()
    if isinstance(context.cash_substitution, dict):
        names = {str(_value(row, 'stock_symbol') or '') for row in rows}
        if set(context.cash_substitution) - names:
            raise ValueError('现金替代选择包含当日清单以外的证券')
    for row in rows:
        symbol = str(_value(row, 'stock_symbol') or '')
        if len(symbol) != 9 or not symbol[:6].isdigit() or symbol[-3:] not in ('.SH', '.SZ'):
            raise ValueError(f'本示例仅准备沪深股票实物篮子：{symbol}')
        if symbol in seen:
            raise ValueError(f'成分券重复：{symbol}')
        seen.add(symbol)
        flag = _integer(_value(row, 'cash_substitution_flag'), '现金替代标志')
        if flag not in (1, 2, 3, 4):
            raise ValueError(f'{symbol} 现金替代标志不支持：{flag}')
        # 必须现金替代、退补现金替代仍走现金，不受False影响。
        selected = (context.cash_substitution.get(symbol, False)
                    if isinstance(context.cash_substitution, dict)
                    else context.cash_substitution)
        if isinstance(context.cash_substitution, dict) and symbol in context.cash_substitution and flag == 3:
            raise ValueError(f'{symbol} 禁止现金替代，请从cash_symbols中移除')
        cash = flag in (2, 4) or (flag == 1 and selected)
        if not cash:
            targets[symbol] = _integer(_value(row, 'quantity'), '成分券数量') * context.baskets
    context.component_symbols.update(seen)
    return targets


def _price(context, data, symbol):
    try:
        bar = data[symbol]
    except KeyError:
        return None
    if bar is None:
        return None
    value = bar.open if context.run_info.matching_type == 1 else bar.close
    try:
        price = float(value)
    except (TypeError, ValueError):
        return None
    return price if isfinite(price) and price > 0 else None


def _track_stock(context, order):
    if order.account != context.account or order.order_book_id not in context.component_symbols:
        return
    key = (order.account, order.order_id)
    if order.status in (-1, 2, 3, 5, 7):
        context.finished.add(key)
        context.pending.pop(key, None)
    elif key not in context.finished:
        context.pending[key] = order.order_book_id


def _prepare(context, data, targets):
    ready = True
    for symbol, required in targets.items():
        account = _account(context)
        position = account.positions.get(symbol)
        held = 0 if position is None else int(position.quantity)
        if held >= required:
            if position.sellable < required:
                ready = False
                _wait(context, f'{symbol} 尚未具备本示例要求的可用数量')
            continue
        ready = False
        if symbol in context.attempted or symbol in context.pending.values():
            continue
        price = _price(context, data, symbol)
        if price is None:
            _wait(context, f'{symbol} 当前无有效价格')
            continue
        # 向上买整手补齐篮子，多出的证券留在账户中。
        quantity = int(ceil((required - held) / 100)) * 100
        if symbol.startswith(('688', '689')):
            quantity = max(200, quantity)
        if account.cash < quantity * price * 1.02 + context.cash_buffer:
            _wait(context, f'{symbol} 补券资金不足，需预留现金替代和费用')
            continue
        context.attempted.add(symbol)
        context.last_buy_date = str(context.trade_date)
        orders = order_shares(
            context.account, symbol, quantity, style=MarketOrderStyle(),
            remark=f'ETF篮子补券_{context.trade_date}',
        )
        if not orders:
            print(f'[{context.trade_date}] {symbol} 补券未返回委托')
        for order in orders or []:
            _track_stock(context, order)
            key = (order.account, order.order_id)
            if order.status in (1, 4) and key in context.pending:
                cancel_order(context.account, order.order_id)
    return ready and not context.pending and context.last_buy_date != str(context.trade_date)


def _track_primary(context, order):
    if order.account != context.account or order.symbol != context.etf:
        return
    key = (order.account, 'PRIMARY', str(order.order_id))
    if key in context.primary_finished:
        return
    if context.primary_action is None:
        return
    if context.primary_key is not None and context.primary_key != key:
        return
    context.primary_key = key
    print(
        f'[申赎回报] 操作={context.primary_action}，申请={order.order_id}，'
        f'份额={order.units}，状态={order.status}，阶段={order.stage}，说明={order.reason}'
    )
    if order.status in (-1, 3):
        context.primary_finished.add(key)
        context.phase = '准备成分券' if context.primary_action == '申购' else '待赎回'
        context.primary_action = None
        context.primary_key = None
    elif order.status == 2 and order.stage == 'SETTLED':
        context.primary_finished.add(key)
        if context.primary_action == '申购':
            context.created_units = int(order.units)
            context.created_date = str(context.trade_date)
            context.phase = '待赎回'
        else:
            context.phase = '完成'
        context.primary_action = None
        context.primary_key = None


def handle_data(context, data):
    today = str(context.trade_date)
    if context.current_date != today:
        context.current_date = today
        context.attempted = set()
        context.wait_messages = set()
    if context.phase == '完成':
        return
    if context.phase == '待交收':
        _wait(context, '已申报，等待实际确认及交收，不重复提交')
        return
    snapshot = _snapshot(context)
    if snapshot is None:
        return
    rule, unit, nav, rows = snapshot

    if context.phase == '买入ETF':
        context.component_symbols.add(context.etf)
        required = unit * context.baskets
        if _prepare(context, data, {context.etf: required}):
            context.created_units = required
            context.phase = '待赎回'
        return
    if context.phase == '准备成分券':
        if _value(rule, 'purchase_allowed_flag') not in (1, '1', True, 'Y'):
            _wait(context, '当日PCF未允许申购')
            return
        targets = _physical_targets(context, rows)
        if not _prepare(context, data, targets):
            return
        if context.primary_attempt_date == today:
            return
        # purchase收金额，不收份额；按当日PCF参考净值换成整篮。
        expected_units = unit * context.baskets
        amount = nav * unit * (context.baskets + 0.00000001)
        if not isfinite(amount) or floor(amount / nav / unit) != context.baskets:
            raise ValueError('申购金额无法准确换算为目标篮数')
        context.primary_attempt_date = today
        context.primary_action = '申购'
        context.primary_key = None
        context.phase = '待交收'
        order = purchase(
            context.account, context.etf, amount,
            cash_substitution=context.cash_substitution,
            remark=f'整篮申购_{today}',
        )
        if order is None:
            raise RuntimeError('申购未返回申请，需核对结果，停止以避免重复提交')
        _track_primary(context, order)
        if order.status != -1 and int(order.units) != expected_units:
            raise RuntimeError('已受理份额与目标不一致，停止后续下单，请核对PCF口径')
    elif context.phase == '待赎回':
        if today == context.created_date or context.primary_attempt_date == today:
            return
        if _value(rule, 'redemption_allowed_flag') not in (1, '1', True, 'Y'):
            _wait(context, '当日PCF未允许赎回')
            return
        if context.created_units <= 0 or context.created_units % unit != 0:
            _wait(context, '申购到账份额不满足当前最小赎回单位，不擅自取整')
            return
        position = _account(context).positions.get(context.etf)
        if position is None or position.sellable < context.created_units:
            _wait(context, '申购到账ETF尚未具备本示例要求的可用数量')
            return
        context.primary_attempt_date = today
        context.primary_action = '赎回'
        context.primary_key = None
        context.phase = '待交收'
        order = redeem(
            context.account, context.etf, context.created_units,
            cash_substitution=False, remark=f'整篮赎回_{today}',
        )
        if order is None:
            raise RuntimeError('赎回未返回申请，需核对结果，停止以避免重复提交')
        _track_primary(context, order)


def on_etf_cr_order_rtn(context, order):
    _track_primary(context, order)


def on_etf_cr_confirm(context, result):
    _track_primary(context, result)


def etf_cr_order_cancel(context, order):
    _track_primary(context, order)


def on_stock_order_rtn(context, order):
    _track_stock(context, order)
    if order.account == context.account and order.order_book_id in context.component_symbols:
        print(
            f'[补券订单回报] {order.order_book_id}，委托={order.order_id}，'
            f'状态={order.status}，已成={order.filled_quantity}，说明={order.message}'
        )


def on_stock_trade_rtn(context, trade):
    if trade.account_id != context.account or trade.contract_code not in context.component_symbols:
        return
    key = (trade.account_id, trade.trade_date, trade.order_id, trade.trade_id)
    if key in context.seen_trades:
        return
    context.seen_trades.add(key)
    print(
        f'[补券成交回报] {trade.contract_code}，数量={trade.volume}，'
        f'价格={trade.price}，费用={trade.cost}'
    )


def stock_order_cancel(context, order):
    _track_stock(context, order)
    if order.account == context.account and order.order_book_id in context.component_symbols:
        print(f'[补券撤单或拒单] {order.order_id}，状态={order.status}，说明={order.message}')


def after_trading(context):
    account = _account(context)
    print(
        f'[{context.trade_date}] 阶段={context.phase}，现金={account.cash:.2f}，'
        f'冻结现金={account.frozen_cash:.2f}，净资产={account.total_value:.2f}'
    )
    for symbol, position in account.positions.items():
        if position.quantity > 0:
            print(f'[持仓] {symbol} 数量={position.quantity}，可卖={position.sellable}')

"""证券两融示例，一轮借入和偿还；股票与ETF共用同一写法。

任务中先启用两融，配置融资额度、可交易标的及融券券源。
credit_mode：融资/融券；repay_method：交易偿还/直接偿还。
融资交易偿还使用卖券还款；融券交易偿还使用买券还券。
直接偿还模式：融资用自有现金还款；融券先买入现券，跨日后直接还券。
从空仓、无信用负债的账户启动。日线或分钟均可，不修改账户配置。
"""

from math import ceil, isfinite
from panda_backtest.api.api import *


STRATEGY_PARAMS = {
    'symbol': '510300.SH',
    'quantity': 1000,
    'credit_mode': '融资',
    'repay_method': '交易偿还',
    'hold_trading_days': 2,
}


def initialize(context):
    context.account = context.run_info.stock_account
    context.symbol = STRATEGY_PARAMS['symbol']
    context.quantity = STRATEGY_PARAMS['quantity']
    context.mode = STRATEGY_PARAMS['credit_mode']
    context.method = STRATEGY_PARAMS['repay_method']
    context.hold_days = STRATEGY_PARAMS['hold_trading_days']
    if context.mode not in ('融资', '融券') or context.method not in ('交易偿还', '直接偿还'):
        raise ValueError('信用方向或偿还方式无效')
    if type(context.quantity) is not int or context.quantity <= 0 or context.quantity % 100:
        raise ValueError('本示例数量使用正的100整数倍')
    if type(context.hold_days) is not int or context.hold_days < 1:
        raise ValueError('持有交易日数必须是正整数')
    context.phase = '借入'
    context.date = None
    context.day_index = 0
    context.hold_start = None
    context.attempt_date = None
    context.stock_buy_date = None
    context.pending = {}
    context.finished = set()
    context.trades = set()
    account = _account(context)
    if any(position.quantity > 0 for position in account.positions.values()):
        raise ValueError('本示例从空仓账户启动')
    snapshot = _credit(context)
    if snapshot['total_debt'] > 0:
        raise ValueError('本示例需要账户没有已有信用负债')
    if context.mode == '融资' and context.run_info.margin_credit_limit <= 0:
        raise ValueError('请在回测任务中设置正的融资额度')


def _account(context):
    account = context.stock_account_dict.get(context.account)
    if account is None:
        raise ValueError('证券账户不存在')
    return account


def _credit(context):
    snapshot = get_margin_snapshot(context.account)
    if not snapshot or not snapshot.get('enabled'):
        raise ValueError('请先在任务设置中启用证券两融')
    return snapshot


def _loan_quantity(context, snapshot):
    record = snapshot['lending'].get(context.symbol)
    return 0 if record is None else int(record['quantity'])


def _track(context, order):
    if order.account != context.account or order.order_book_id != context.symbol:
        return
    key = (order.account, order.order_id)
    if order.status in (-1, 2, 3, 5, 7):
        context.finished.add(key)
        context.pending.pop(key, None)
    elif key not in context.finished:
        context.pending[key] = order.order_id


def _accept_orders(context, orders):
    for order in orders or []:
        _track(context, order)
        if order.status in (1, 4, 6) and (order.account, order.order_id) in context.pending:
            cancel_order(context.account, order.order_id)


def _price(context, data):
    try:
        bar = data[context.symbol]
    except KeyError:
        return None
    if bar is None:
        return None
    value = bar.open if context.run_info.matching_type == 1 else bar.close
    if value is None:
        return None
    price = float(value)
    return price if isfinite(price) and price > 0 else None


def handle_data(context, data):
    today = str(context.trade_date)
    if today != context.date:
        context.date = today
        context.day_index += 1
    if context.phase == '完成':
        return
    snapshot = _credit(context)
    if snapshot['risk_state'] == 'LIQUIDATION':
        # 不与自动强平同时追加买卖，后续继续观察真实负债。
        context.phase = '偿还'
        return
    if context.pending:
        return
    if context.phase == '借入':
        borrowed = (snapshot['financing_balance'] > 0 if context.mode == '融资'
                    else _loan_quantity(context, snapshot) > 0)
        if borrowed:
            context.phase = '持有'
            context.hold_start = context.day_index
        elif context.attempt_date != today and snapshot['risk_state'] == 'NORMAL':
            if _price(context, data) is None:
                return
            context.attempt_date = today
            if context.mode == '融资':
                orders = financing_buy(context.account, context.symbol, context.quantity,
                                       style=MarketOrderStyle(), remark='融资示例')
            else:
                orders = securities_lending_sell(context.account, context.symbol, context.quantity,
                                                style=MarketOrderStyle(), remark='融券示例')
            _accept_orders(context, orders)
            snapshot = _credit(context)
            borrowed = (snapshot['financing_balance'] > 0 if context.mode == '融资'
                        else _loan_quantity(context, snapshot) > 0)
            if borrowed:
                context.phase = '持有'
                context.hold_start = context.day_index
            return
        else:
            return
    if context.phase == '持有':
        if context.day_index - context.hold_start < context.hold_days:
            return
        context.phase = '偿还'
    if snapshot['total_debt'] <= 0:
        context.phase = '完成'
        print(f'[{today}] 信用负债已清偿；现券持仓保持不动')
        return
    if context.attempt_date == today:
        return
    account = _account(context)
    position = account.positions.get(context.symbol)
    sellable = 0 if position is None else int(position.sellable)
    context.attempt_date = today

    if context.mode == '融资':
        debt = float(snapshot['financing_balance']) + float(snapshot['financing_interest'])
        if context.method == '交易偿还' and sellable > 0:
            if _price(context, data) is None:
                context.attempt_date = None
                return
            _accept_orders(context, sell_repayment(
                context.account, context.symbol, sellable,
                style=MarketOrderStyle(), remark='卖券还款示例',
            ))
        elif context.method == '直接偿还' or position is None or position.quantity <= 0:
            amount = min(float(account.cash), debt)
            if amount > 0:
                actual = repay_financing(context.account, amount)
                print(f'[直接还款] 实际归还={actual}')
    else:
        quantity = _loan_quantity(context, snapshot)
        if quantity > 0 and context.method == '交易偿还':
            if _price(context, data) is None:
                context.attempt_date = None
                return
            _accept_orders(context, buy_to_cover(
                context.account, context.symbol, quantity,
                style=MarketOrderStyle(), remark='买券还券示例',
            ))
        elif quantity > 0:
            if sellable >= quantity and context.stock_buy_date != today:
                actual = return_securities(context.account, context.symbol, quantity)
                print(f'[直接还券] 实际归还={actual}')
            else:
                held = 0 if position is None else int(position.quantity)
                missing = quantity - held
                price = _price(context, data)
                if missing > 0 and price is not None:
                    buy_quantity = int(ceil(missing / 100)) * 100
                    if context.symbol.startswith(('688', '689')):
                        buy_quantity = max(200, buy_quantity)
                    if account.cash >= buy_quantity * price * 1.02 + 100:
                        context.stock_buy_date = today
                        _accept_orders(context, order_shares(
                            context.account, context.symbol, buy_quantity,
                            style=MarketOrderStyle(), remark='为直接还券买入现券',
                        ))
    if not context.pending and _credit(context)['total_debt'] <= 0:
        context.phase = '完成'
        print(f'[{today}] 信用负债已清偿；现券持仓保持不动')


def on_stock_order_rtn(context, order):
    _track(context, order)
    if order.account == context.account and order.order_book_id == context.symbol:
        print(f'[订单] {order.order_id} 状态={order.status} 用途={order.margin_usage} '
              f'强平={order.force_liquidation} 说明={order.message}')


def on_stock_trade_rtn(context, trade):
    if trade.account_id != context.account or trade.contract_code != context.symbol:
        return
    key = (trade.account_id, trade.trade_date, trade.order_id, trade.trade_id)
    if key in context.trades:
        return
    context.trades.add(key)
    print(f'[成交] {trade.contract_code} 数量={trade.volume} 价格={trade.price} '
          f'用途={trade.margin_usage} 强平={trade.force_liquidation} 费用={trade.cost}')


def stock_order_cancel(context, order):
    _track(context, order)
    if order.account == context.account and order.order_book_id == context.symbol:
        print(f'[撤单或拒单] {order.order_id} 状态={order.status} 说明={order.message}')


def after_trading(context):
    snapshot = _credit(context)
    print(f'[{context.trade_date}] 阶段={context.phase}，总负债={snapshot["total_debt"]}，'
          f'融资={snapshot["financing_balance"]}，融券数量={_loan_quantity(context, snapshot)}，'
          f'风险状态={snapshot["risk_state"]}')